Cqrs Es Bundle Laravel Package

Getting Started

Minimal Setup

1. Installation

   composer require averor/cqrs-es-bundle
   

   Add to config/bundles.php:

   return [
       // ...
       Averor\CqrsEsBundle\AverorCqrsEsBundle::class => ['all' => true],
   ];
   

2. Basic Configuration Edit config/packages/averor_cqrs_es.yml to define:

   • message_bus (e.g., symfony/messenger or php-amqplib)
   • event_store (e.g., spiral/event-sourced or eventSauce/event-sourced)
   • serializer (default: Symfony’s Serializer)

3. First Use Case: Dispatching a Command

   use Averor\CqrsEsBundle\MessageBus\CommandBusInterface;
   
   class CreateUserCommandHandler implements CommandHandlerInterface {
       public function __invoke(CreateUserCommand $command) {
           // Handle command logic
       }
   }
   
   // In a controller/service:
   $commandBus = $this->container->get(CommandBusInterface::class);
   $commandBus->dispatch(new CreateUserCommand());
   

4. First Use Case: Publishing an Event

   use Averor\CqrsEsBundle\EventStore\EventStoreInterface;
   
   $eventStore = $this->container->get(EventStoreInterface::class);
   $eventStore->appendTo('user-123', new UserCreatedEvent());
   

Implementation Patterns

Core Workflows

1. Command-Query-Responsibility Segregation (CQRS)

• Commands: Use CommandBusInterface for write operations.

  $commandBus->dispatch(new UpdateUserCommand($userId, ['name' => 'Alice']));
  

• Queries: Use QueryBusInterface for read operations (if integrated with a query bus like league/tactician).

  $queryBus->ask(new GetUserQuery($userId));
  

2. Event Sourcing

• Aggregate Roots: Implement AggregateRoot interface.

  class UserAggregate implements AggregateRoot {
      private $id;
      private $name;
  
      public function handle(CreateUserCommand $command) {
          $this->apply(new UserCreatedEvent($command->name));
      }
  
      public function apply(UserCreatedEvent $event) {
          $this->name = $event->name;
      }
  }
  

• Event Store Integration:

  $eventStore->load('user-123', UserAggregate::class); // Rehydrates aggregate
  $eventStore->persist($aggregate); // Saves events
  

3. Middleware Pipelines

• Command/Event Middleware: Extend functionality via middleware.

  # config/packages/averor_cqrs_es.yml
  averor_cqrs_es:
      command_bus:
          middleware: ['averor_cqrs_es.middleware.command.logger']
  

  Create middleware:

  class CommandLoggerMiddleware implements CommandMiddlewareInterface {
      public function handle(CommandMessage $command, callable $next) {
          // Pre-processing
          $result = $next($command);
          // Post-processing
          return $result;
      }
  }
  

4. Integration with Symfony Messenger

• Bridge commands/events to Symfony Messenger:

  averor_cqrs_es:
      message_bus:
          command_bus: messenger
          event_bus: messenger
  

  Configure Messenger transports in config/packages/messenger.yaml.

Best Practices

Dependency Injection

• Prefer constructor injection for CommandBusInterface, EventStoreInterface, and QueryBusInterface.

  public function __construct(
      private CommandBusInterface $commandBus,
      private EventStoreInterface $eventStore
  ) {}
  

Testing

• Mocking the Bus/Store:

  $mockBus = $this->createMock(CommandBusInterface::class);
  $this->container->set(CommandBusInterface::class, $mockBus);
  

• Event Store Snapshots: Use EventStore::load() with snapshots for performance.

  $eventStore->load('user-123', UserAggregate::class, 100); // Load from snapshot at version 100
  

Performance

• Batch Processing: For bulk commands/events, use dispatchMany() or appendMany().

  $commandBus->dispatchMany([new Command1(), new Command2()]);
  

Gotchas and Tips

Pitfalls

1. Circular Dependencies

• Issue: Event handlers subscribed to events emitted by the same aggregate can cause infinite loops.
• Fix: Use EventStore::appendTo() with explicit event filtering or middleware to validate event sources.

2. Serialization Errors

• Issue: Custom events/commands may not serialize/deserialize correctly.
• Fix: Implement Serializable or configure the serializer in averor_cqrs_es.yml:

  averor_cqrs_es:
      serializer:
          format: 'json'
          context: { groups: ['serializable'] }
  

3. Transaction Management

• Issue: Event persistence may fail mid-transaction.
• Fix: Wrap in a transaction (Doctrine DBAL or Symfony’s Transaction component):

  $connection->beginTransaction();
  try {
      $eventStore->appendTo($aggregateId, $event);
      $connection->commit();
  } catch (\Exception $e) {
      $connection->rollBack();
      throw $e;
  }
  

4. Middleware Order

• Issue: Middleware execution order may not be intuitive.
• Fix: Explicitly define order in config:

  averor_cqrs_es:
      command_bus:
          middleware:
              - averor_cqrs_es.middleware.command.validation
              - averor_cqrs_es.middleware.command.logger
  

Debugging Tips

1. Logging

• Enable debug logging for the bundle:

  averor_cqrs_es:
      debug: true
  

• Check logs for dispatched commands/events:

  bin/console debug:container averor_cqrs_es.logger
  

2. Event Store Inspection

• List all events for an aggregate:

  $events = $eventStore->getEvents('user-123');
  

• Replay events manually:

  $aggregate = new UserAggregate();
  foreach ($events as $event) {
      $aggregate->apply($event);
  }
  

3. Command/Event Validation

• Use middleware for validation:

  class ValidateCommandMiddleware implements CommandMiddlewareInterface {
      public function handle(CommandMessage $command, callable $next) {
          if (!$command->isValid()) {
              throw new \InvalidArgumentException('Invalid command');
          }
          return $next($command);
      }
  }
  

Extension Points

1. Custom Event Stores

• Implement EventStoreInterface for alternative backends (e.g., MongoDB, PostgreSQL):

  class MongoEventStore implements EventStoreInterface {
      public function appendTo(string $aggregateId, EventInterface $event) {
          // Custom logic
      }
  }
  

• Register in config:

  averor_cqrs_es:
      event_store: averor_cqrs_es.event_store.mongo
  

2. Custom Message Buses

• Extend MessageBusInterface for non-Symfony/Messenger buses (e.g., RabbitMQ):

  class RabbitMqBus implements MessageBusInterface {
      public function dispatch(MessageInterface $message) {
          // RabbitMQ logic
      }
  }
  

3. Domain Events

• Publish domain events after commands:

  $commandBus->dispatch(new CreateUserCommand(), [
      new PublishDomainEventMiddleware(new UserCreatedEvent())
  ]);
  

4. Projections

• Use event listeners to update projections (e.g., read models):

  $eventStore->listen(UserCreatedEvent::class, function (UserCreatedEvent $event) {
      // Update read model
  });