The problem you've encountered has a lot to do with messaging implementation.

The solution of the problem you're describing depends on many variables and it is very difficult to reliably take into account all of them and build a generally-applicable robust solution. In fact, the problem is so difficult to tackle, people like Udi Dahan have created entire companies to build a reliable solution for the problem and to provide the solution as a service.

The simplest version of the problem is: when a certain event happens within the system, I want to react to it in the same database transaction (i.e. also synchronously) and do some synchronous operation as a reaction. This situation is fairly common, when you have a project containing multiple bounded contexts, you want to do some operation on a certain aggregate in a context B when an event on a different aggregate happens in a context A but still want to retain the atomicity that either N operations will succeed or none. Thanks to this mechanism, the bounded contexts are not directly tied to each other which proves to be a great thing when you decide to split the BCs between multiple projects and/or applications.

Implementing this solution is as easy as having a single TransactionManager spanning your entire application and building a simple EventDispatcher class with a dispatch(event: Message): void method, which internally loops over all attached event handlers as listeners to the dispatched event passes the dispatched event to the handler as a parameter.

Unfortunately, things get much more complicated very quickly, once you start working with multiple threads, want to process events asynchronously and/or have bounded contexts as separate micro services, among which keeping transactional safety is pretty much impossible. Along with eventual consistency, this introduces another requirements your application now must fulfil in order for its messaging processes to be considered reliable.

Whether it's simple asynchronous processing within the same application or the asynchronous processing is handled by dispatching the call to another micro-service, you'll need to start tracking a lot of information about what's happening in your system(s) in a transactionally safe way.

#An effort to try to implement a transactionally-safe message handling

^{Let's have a look at the modelling process, starting with the simplest solution and gradually move onto better and better implementation.}

You start with a simple MyEventHandler, handling MyEvent, which depends on an AsynchronousServiceBus and calls serviceBus.dispatch(new SomeCommand()) as a response to a dispatched MyEvent in your system. What the AsynchronousServiceBus service bus does is, it takes the message passed to the dispatch method and forwards it to RabbitMQ from which it is asynchronously consumed and processed by a worker.

You program your system in the following way:

1. a command is issued to create an entity,
2. you start a database transaction to make the creation transactionally safe,
3. before committing the transaction, you dispatch a MyEvent to the system,
4. committing the transaction fails (e.g. due to unavailability of the database, unique check violation,...).

You've just ended up in a situation, where there's an event flying through your system which shouldn't have been present at all, because the entity has not been completed. To fix this problem, you reprogram your application to dispatch events after you commit a database transaction. You're happy with your solution, because now when an event is fired by your application, you can be sure it makes sense to do it so. Unfortunately, soon enough you encounter a problematic situation which started happening quite often:

1. a command is issued to create an entity,
2. you start a database transaction,
3. you commit the transaction and it goes through without a problem,
4. you try to dispatch the event after commit, but RabbitMQ fails to respond.

Because of 4th step, the event which had to be dispatched got lost and your commited entity from 3rd step is in a waiting state and will never be completed, because there's no record of the event in your system.

You figure out the only way how to fix the problem you've just encountered is to store dispatched events in the same database using the same transaction as storing the affected entity. Because of that, you reprogram your system once again in the following way:

1. a command is issued to create an entity,
2. you start a database transaction,
3. instead of directly handling an event by sending a command to RabbitMQ, the event is now persisted to a new dispatched_event database table using an INSERT statement in a PREPARED state,
4. you commit the transaction and it goes through without a problem,
5. polling service checks the dispatched_event for PREPARED events, loads those and pushes them over to RabbitMQ one by one,
6. when RabbitMQ responds with message accepted, the polling service changes the state of the event to SENT.

Unfortunately what has happened, the polling service has started to fail from time to time between successfully sending the message to RabbitMQ and changing the message's state to SENT, and you find out a lot of duplicated messages have started to appear in your infrastructure. In order to fix this problem, you introduce a unique id in the form of GUID and append this attribute to your message. You also adapt the RabbitMQ workers to check whether a message with a specified id has been already processed or not (this is done by the worker storing the ids of messages which have already been processed), and your problem with duplicated messages is solved.

Another developer joins your project and you find out that when MyEvent is dispatched, two followup commands are supposed to be executed. Because your network connection is really unstable, what happens from time to time, the first of the two commands is dispatched to RabbitMQ and successfully processed, but the second never went through. Because of this, you have once again ended up in a situation where only a partial change has been applied to your system and your system starts to be out of sync. Unfortunately, thanks to your deduplication logic, your commands are sharing the same event id and are discarded as duplicates on retry and never repeated.

In order to fix this problem, you realise what you really need to store is commands and not events, so you remodel your dispatched_event table to dispatched_command, rework your polling service and now effectively store all commands to be dispatched as a reaction to a specific event within your system. Thanks to this new implementation, specific failing commands can now be retried multiple times.

Introducing other constraints, such as having messages to be sent at most once or messages to be sent at least once introduces further complexity and constraints on your system and further complicates the potential solution.

Since you're in a PHP, I assume what you're after is the initial-simple synchronous implementation. That alone is fairly simple to do, should you have a decent transaction manager ensuring a transaction spans across the desired change to your system as well as changes invoked by listeners attached to dispatched events. A concrete solution for asynchronous messaging is a topic for complex discussion and also depends on your business needs and a general solution cannot really be provided.