[EventsSubscription] Fix transaction scope in EventsSubscription

[hkawalek]

Problem description

When EventsSubscription processes event, it begins new transaction and then starts executing registered onEvent
callbacks. Then sample onEvent callback publishes command which starts flow CommandHandler -> ApplicationService -> EventRepository.read -> DomainLogic -> EventRepository.write -> EventBus. During EventRepository.write step, the new different transaction begins and commits successfully. If after this transaction next onEvent throws error, the transaction which was started by EventsSubscription is aborted but the changes made by EventRepository.write heve already been persisted. Consequently EventsSubscription is in inconsistent state because currentPosition hasn't changed but some changes of onEvent occured.

Solution

EventRepository.write executed by onEvent callback should use transaction started in EventsSubscription.

Summon: @nowakprojects

[github-actions]

Branch issue-360-_EventsSubscription_Fix_transaction_scope_in_EventsSubscription created!

[MateuszNaKodach]

What do you think about introducing some abstraction over Prisma's transactions?
Something like pattern UnitOfWork?
Do you have any idea how to do it with implementation like PrismaTransactionUnitOfWork :) ?

[hkawalek]

Using UnitOfWork will be great but it will be very hard to implement because of two reasons:

1. Dependency injection support in nestjs is very poor. We can't create lifetime scopes and REQUEST doesn't solve our problem because its lifetime is bounded with HTTP request (we need to create scope in EventsSubscription).
2. Implementing entity tracker over prisma client (which isn't stable - it's autogenerated) would be too complicated.

I think that we don't need full implementation of UnitOfWork. We only want every EventRepository.write call would be
included in the same transaction when EventsSubscription is proccessing event. I have a little bit hacky idea to inject transaction context into command when it's send over commandBus and then dispatch that context inside ApplicationService.execute and then create TransactionEventRepository which will save changes to currently received transaction context. Maybe more investigation is needed?