Proposal: Spring-like TransactionalEventListener/TransactionalEventPublisher but for @nestjs/cqrs

[frct1]

Hi,
We using nestjs-cls for all over the places of our backend and @transactional really solved tx propagation on request pipeline. As we actively using @nestjs/cqrs I guess there could be an option as Spring offer like TransactionalEventPublisher that binds to current tx execution and postpone event publish until commit.

Is there any options to do the same thing for nestjs-cls/transactional?
Probably it could be something like Observable that receives callbacks that fire events in them same order as received.

  // When transaction initialised we create ReplaySubject as storage for all events.
  const afterCommitEvents = new ReplaySubject<DomainEvent>();

  // During transaction - events buffered in order
  this.txHost.tx.afterCommitEvents.next(new ProjectCreatedEvent(...));
  this.txHost.tx.afterCommitEvents.next(new WalletCreatedEvent(...));

  // Once transaction commited and connection released proceeding all stored events.
  afterCommitEvents.subscribe(event => this.eventBus.publish(event));
  afterCommitEvents.complete();

Pros:

• Confidence that events are triggered when transaction succeed

Cons:

• nestjs-cls/transactional will depend on @nestjs/cqrs.

Comments:
Not sure how we can avoid dependency on cqrs package, but it could be just a feature requirement tbh and we can go with that.

[Papooch]

Thanks for the suggestion, but please see this discussion thread first as it might be related: #221. This issue might also be related: #176

I don't like this idea for the following reason:
There is no guarantee that events emitted in memory will ever be processed at all, since the server can crash at any time. Mixing this approach with transactions could make the app's data very easily end up in an inconsistent state.

The only way to reliably process events in reaction to state change is to use the "Outbox pattern" and save the event to the database within the same transaction.
Without that, anything that happens after the transaction is committed or rolled could fail independently of the state change. That's one of the reasons I dislike the Saga implementation of nestjs/cqrs, because it is completely oblivious to this hidden issue.

---

That said, what you propose might me achievable by creating a custom transactional adapter and baking this behavior into the wrapWithTransaction logic, while also providing a custom tx object, that exposes a method for aggregating the events.

[frct1]

That's a good point

@nestjs/cqrs has it's culprits for sure, anyway since there is no transaction-first practices we have to deal with what we have.

As a workaround for TypeOrm we came up to use it's EntitySubscriber feature + nestjs-cls. However it is not as wide applicable for other adapters if there is no such feature to trigger events on transaction pipeline as Typeorm have. (1)

Storing events to be published in database is a really good idea but there is no easy option to immediately trigger events as they added to database (kinda fifo queue).

(1) Not speaking in terms of triggering cqrs events using transactional plugin, but what do you think on hooks/events feature from transactional like TypeOrm has. This way transactional plugin might become a nestjs standard way on implementing transactions.

Spring sample I played with 2 years ago:

@EventListener
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) // Event triggers only after commit
public void handle(AddFundsAttemptEvent event) throws JsonProcessingException {
  Customer customer = event.getCustomer();
  Payment payment = event.getPayment();
  HashMap<String, Object> data = new HashMap<>();
  data.put("customer", customerMapper.toDto(customer));
  data.put("payment", paymentMapper.toDto(payment));
  this.natsService.publish(event.getEventName(), data);
}

[Papooch]

Storing events to be published in database is a really good idea but there is no easy option to immediately trigger events as they added to database (kinda fifo queue).

When working with an oubox, I've adopted a simple technique: Once the events are added to the transactional outbox and the transaction succeeds, I explicitly trigger the outbox publisher to process all outbox messages once the request finishes. Otherwise the outbox publishers polls for new outbox messages every second (or any other amount of time). Most databases also support change streams, so you can immediately trigger the publisher once the database emits a new insert to the outbox table/collection.

---

The "problem" with Nest is that there's no standard data API like there is JPA in Java. Each and every database library comes with its custom data and transaction API, so Nest would have to provide adapters for all popular libraries, basically what this library does now, but on a much greater scale. I sometimes think about creating new plugins for these kinds of behaviors for nestjs-cls, but then get discouraged when I remember the maintenance burden that this would create for the furure.

---

However, I looked into it more and it seems that provided we added support for injecting arbitrary providers into the adapter, which should be straightforward change (edit: this is possible since the transactional plugin's version 3.2.0), it should be possible to create a custom adapter that implements what you propose (even with the outbox support). I just don't think there's such benefit in creating a general solution, since there is no standard API to lean on on Nest's side.