Message Repository For Doctrine Laravel Package

Getting Started

Minimal Setup

1. Install the Package

   composer require eventsauce/message-repository-for-doctrine
   

2. Configure Doctrine DBAL Connection Ensure you have a Doctrine DBAL connection configured in your Laravel app (e.g., via config/database.php or a custom connection).

3. Define the Message Repository

   use EventSauce\MessageRepository\DoctrineMessageRepository\DoctrineUuidV4MessageRepository;
   use EventSauce\MessageRepository\TableSchema\DefaultTableSchema;
   use EventSauce\UuidEncoding\BinaryUuidEncoder;
   use EventSauce\EventSauce;
   
   // Initialize the repository
   $messageRepository = new DoctrineUuidV4MessageRepository(
       connection: $doctrineDbalConnection,
       tableName: 'event_store',
       serializer: EventSauce::serializer(),
       tableSchema: new DefaultTableSchema(),
       uuidEncoder: new BinaryUuidEncoder(),
   );
   

4. Create the Database Table Run a migration to create the table matching the schema (e.g., event_id, aggregate_root_id, version, payload). Example migration:

   Schema::create('event_store', function (Blueprint $table) {
       $table->binaryUuid('event_id')->primary();
       $table->binaryUuid('aggregate_root_id');
       $table->unsignedInteger('version');
       $table->text('payload');
       $table->timestamps(); // Optional: for debugging or auditing
   });
   

5. First Use Case: Storing Events

   $event = new SomeEvent('data');
   $messageRepository->store($event, new UuidV4(), 1);
   

Implementation Patterns

Core Workflows

1. Event Storage Store events for an aggregate root:

   $messageRepository->store($event, $aggregateRootId, $version);
   

2. Loading Events Load events for an aggregate root:

   $events = $messageRepository->load($aggregateRootId);
   

3. Appending Events Append events to an aggregate root (atomic operation):

   $messageRepository->append($aggregateRootId, $events);
   

4. Custom Table Schema Extend DefaultTableSchema for custom columns:

   use EventSauce\MessageRepository\TableSchema\TableSchema;
   
   class CustomTableSchema extends TableSchema {
       public function additionalColumns(): array {
           return [
               new Header('metadata', 'json'),
           ];
       }
   }
   

Integration with Laravel

1. Service Provider Binding Bind the repository in a service provider:

   $this->app->bind(DoctrineUuidV4MessageRepository::class, function ($app) {
       return new DoctrineUuidV4MessageRepository(
           $app->make(Connection::class),
           'event_store',
           EventSauce::serializer(),
           new DefaultTableSchema(),
           new BinaryUuidEncoder(),
       );
   });
   

2. Event Sourcing Pattern Use the repository in an event-sourced aggregate root:

   class Order {
       private $events = [];
   
       public function apply(OrderEvent $event) {
           $this->events[] = $event;
           // Apply event logic
       }
   
       public function getUncommittedEvents() {
           return $this->events;
       }
   
       public function persist() {
           $messageRepository->append($this->id, $this->getUncommittedEvents());
           $this->events = [];
       }
   }
   

3. Transaction Management Wrap repository operations in a DBAL transaction:

   $connection->beginTransaction();
   try {
       $messageRepository->append($aggregateRootId, $events);
       $connection->commit();
   } catch (\Exception $e) {
       $connection->rollBack();
       throw $e;
   }
   

4. Event Deserialization Customize deserialization by extending the serializer or using middleware:

   $serializer = EventSauce::serializer();
   $serializer->addMiddleware(new CustomDeserializationMiddleware());
   

Gotchas and Tips

Pitfalls

1. UUID Type Mismatch

   • If your database lacks a native uuid type, use BinaryUuidEncoder to avoid storage issues.
   • If using UuidStringEncoder, ensure the database column is text or varchar with sufficient length (e.g., 36 for UUIDv4).

2. Table Schema Mismatch

   • Mismatched column names (e.g., aggregate_root_version vs. version) will cause queries to fail.
   • Always verify the table schema matches the TableSchema implementation.

3. Concurrency Issues

   • The repository does not handle optimistic concurrency by default. Use version checks in your application logic:

     $currentVersion = $messageRepository->getVersion($aggregateRootId);
     if ($currentVersion !== $expectedVersion) {
         throw new ConcurrencyException();
     }
     

4. Performance with Large Event Streams

   • Loading all events for a long-lived aggregate root can be slow. Consider:
     • Pagination (e.g., load events in batches).
     • Projections or read models for common queries.

5. Serialization Errors

   • Ensure all events implement EventSauce\EventSauce interfaces (e.g., EventSauce\EventSauce).
   • Custom serializers may fail if events contain non-serializable properties (e.g., closures, resources).

Debugging Tips

1. Enable Doctrine DBAL Logging Configure DBAL to log SQL queries for debugging:

   $connection->getConfiguration()->setSQLLogger(new \Doctrine\DBAL\Logging\EchoSQLLogger());
   

2. Check for Silent Failures

   • The repository may silently fail if the table or columns are missing. Validate the schema:

     $schemaManager = $connection->createSchemaManager();
     if (!$schemaManager->tablesExist(['event_store'])) {
         throw new \RuntimeException('Event store table does not exist');
     }
     

3. UUID Generation

   • Ensure UuidV4 is used consistently for event_id and aggregate_root_id to avoid type conflicts.

Extension Points

1. Custom Headers Add metadata to events via additionalColumns in a custom TableSchema:

   class MetadataTableSchema extends TableSchema {
       public function additionalColumns(): array {
           return [
               new Header('created_at', 'datetime'),
               new Header('user_id', 'text'),
           ];
       }
   }
   

2. Event Filtering Extend the repository to filter events by metadata:

   class FilteredMessageRepository extends DoctrineUuidV4MessageRepository {
       public function loadWithMetadata($aggregateRootId, $metadata) {
           $query = $this->createSelectQuery($aggregateRootId);
           $query->andWhere('metadata = :metadata');
           $query->setParameter('metadata', json_encode($metadata));
           return $this->loadFromQuery($query);
       }
   }
   

3. Bulk Operations Optimize bulk inserts by batching events:

   $connection->beginTransaction();
   foreach ($events in array_chunk($allEvents, 100)) {
       $messageRepository->append($aggregateRootId, $events);
   }
   $connection->commit();
   

4. Read Replicas For read-heavy workloads, configure a read replica connection:

   $readConnection = $connection->getWrappedConnection()->getReadConnection();
   $readRepository = new DoctrineUuidV4MessageRepository(
       $readConnection,
       'event_store',
       $serializer,
       $tableSchema,
       $uuidEncoder,
   );
   

5. Event Projections Use the repository to build projections (e.g., materialized views):

   $events = $messageRepository->load($aggregateRootId);
   $projection = new OrderProjection();
   foreach ($events as $event) {
       $projection->apply($event);
   }