Repository: patchlevel/event-sourcing (php)

Issue:
MultiTable pipeline support
Currently only the SingleTableStore works with the pipeline. With MultiTableStore you have the problem of getting all events from different tables and putting them in the correct order.

Solutions:
1) Save microseconds as well (not supported by every DB) in order to sort according to them
2) Add a meta table by storing the meta information of each event. Autoincrement, Aggregate, AggregateId etc.

I currently tend to variant 2.

Pull request:
multi table pipeline support
fix #40 

Patch:
diff --git a/src/Console/ProjectionDropCommand.php b/src/Console/ProjectionDropCommand.php
index ee6dbdd1..eac51d82 100644
--- a/src/Console/ProjectionDropCommand.php
+++ b/src/Console/ProjectionDropCommand.php
@@ -27,7 +27,7 @@ class ProjectionDropCommand extends Command
 
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
-        $this->projectionRepository->drop();
+        $this->projectionRepository->create();
 
         return 0;
     }
diff --git a/src/Pipeline/Middleware/ChainMiddleware.php b/src/Pipeline/Middleware/ChainMiddleware.php
deleted file mode 100644
index 2e5ef327..00000000
--- a/src/Pipeline/Middleware/ChainMiddleware.php
+++ /dev/null
@@ -1,51 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Patchlevel\EventSourcing\Pipeline\Middleware;
-
-use Patchlevel\EventSourcing\Pipeline\EventBucket;
-
-class ChainMiddleware implements Middleware
-{
-    /** @var list<Middleware> */
-    private array $middlewares;
-
-    /**
-     * @param list<Middleware> $middlewares
-     */
-    public function __construct(array $middlewares)
-    {
-        $this->middlewares = $middlewares;
-    }
-
-    /**
-     * @return list<EventBucket>
-     */
-    public function __invoke(EventBucket $bucket): array
-    {
-        $buckets = [$bucket];
-
-        foreach ($this->middlewares as $middleware) {
-            $buckets = $this->processMiddleware($middleware, $buckets);
-        }
-
-        return $buckets;
-    }
-
-    /**
-     * @param list<EventBucket> $buckets
-     *
-     * @return list<EventBucket>
-     */
-    private function processMiddleware(Middleware $middleware, array $buckets): array
-    {
-        $result = [];
-
-        foreach ($buckets as $bucket) {
-            $result += $middleware($bucket);
-        }
-
-        return $result;
-    }
-}
diff --git a/src/Pipeline/Middleware/ClassRenameMiddleware.php b/src/Pipeline/Middleware/ClassRenameMiddleware.php
deleted file mode 100644
index 5d91d305..00000000
--- a/src/Pipeline/Middleware/ClassRenameMiddleware.php
+++ /dev/null
@@ -1,50 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Patchlevel\EventSourcing\Pipeline\Middleware;
-
-use Patchlevel\EventSourcing\Aggregate\AggregateChanged;
-use Patchlevel\EventSourcing\Pipeline\EventBucket;
-
-use function array_key_exists;
-use function get_class;
-
-class ClassRenameMiddleware implements Middleware
-{
-    /** @var array<class-string<AggregateChanged>, class-string<AggregateChanged>> */
-    private array $classes;
-
-    /**
-     * @param array<class-string<AggregateChanged>, class-string<AggregateChanged>> $classes
-     */
-    public function __construct(array $classes)
-    {
-        $this->classes = $classes;
-    }
-
-    /**
-     * @return list<EventBucket>
-     */
-    public function __invoke(EventBucket $bucket): array
-    {
-        $event = $bucket->event();
-        $class = get_class($event);
-
-        if (!array_key_exists($class, $this->classes)) {
-            return [$bucket];
-        }
-
-        $data = $event->serialize();
-        $data['event'] = $this->classes[$class];
-
-        $newEvent = AggregateChanged::deserialize($data);
-
-        return [
-            new EventBucket(
-                $bucket->aggregateClass(),
-                $newEvent
-            ),
-        ];
-    }
-}
diff --git a/src/Pipeline/Middleware/ExcludeEventMiddleware.php b/src/Pipeline/Middleware/DeleteEventMiddleware.php
similarity index 93%
rename from src/Pipeline/Middleware/ExcludeEventMiddleware.php
rename to src/Pipeline/Middleware/DeleteEventMiddleware.php
index 491676a4..19f97d80 100644
--- a/src/Pipeline/Middleware/ExcludeEventMiddleware.php
+++ b/src/Pipeline/Middleware/DeleteEventMiddleware.php
@@ -7,7 +7,7 @@ namespace Patchlevel\EventSourcing\Pipeline\Middleware;
 use Patchlevel\EventSourcing\Aggregate\AggregateChanged;
 use Patchlevel\EventSourcing\Pipeline\EventBucket;
 
-class ExcludeEventMiddleware implements Middleware
+class DeleteEventMiddleware implements Middleware
 {
     /** @var list<class-string<AggregateChanged>> */
     private array $classes;
diff --git a/src/Pipeline/Middleware/FilterEventMiddleware.php b/src/Pipeline/Middleware/FilterEventMiddleware.php
index 67da8f8a..426bd851 100644
--- a/src/Pipeline/Middleware/FilterEventMiddleware.php
+++ b/src/Pipeline/Middleware/FilterEventMiddleware.php
@@ -9,15 +9,15 @@ use Patchlevel\EventSourcing\Pipeline\EventBucket;
 
 class FilterEventMiddleware implements Middleware
 {
-    /** @var callable(AggregateChanged $event):bool */
-    private $callable;
+    /** @var list<class-string<AggregateChanged>> */
+    private array $classes;
 
     /**
-     * @param callable(AggregateChanged $event):bool $callable
+     * @param list<class-string<AggregateChanged>> $classes
      */
-    public function __construct(callable $callable)
+    public function __construct(array $classes)
     {
-        $this->callable = $callable;
+        $this->classes = $classes;
     }
 
     /**
@@ -25,10 +25,10 @@ class FilterEventMiddleware implements Middleware
      */
     public function __invoke(EventBucket $bucket): array
     {
-        $result = ($this->callable)($bucket->event());
-
-        if ($result) {
-            return [$bucket];
+        foreach ($this->classes as $class) {
+            if ($bucket->event() instanceof $class) {
+                return [$bucket];
+            }
         }
 
         return [];
diff --git a/src/Pipeline/Middleware/IncludeEventMiddleware.php b/src/Pipeline/Middleware/IncludeEventMiddleware.php
deleted file mode 100644
index 20811de1..00000000
--- a/src/Pipeline/Middleware/IncludeEventMiddleware.php
+++ /dev/null
@@ -1,36 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Patchlevel\EventSourcing\Pipeline\Middleware;
-
-use Patchlevel\EventSourcing\Aggregate\AggregateChanged;
-use Patchlevel\EventSourcing\Pipeline\EventBucket;
-
-class IncludeEventMiddleware implements Middleware
-{
-    /** @var list<class-string<AggregateChanged>> */
-    private array $classes;
-
-    /**
-     * @param list<class-string<AggregateChanged>> $classes
-     */
-    public function __construct(array $classes)
-    {
-        $this->classes = $classes;
-    }
-
-    /**
-     * @return list<EventBucket>
-     */
-    public function __invoke(EventBucket $bucket): array
-    {
-        foreach ($this->classes as $class) {
-            if ($bucket->event() instanceof $class) {
-                return [$bucket];
-            }
-        }
-
-        return [];
-    }
-}
diff --git a/src/Pipeline/Pipeline.php b/src/Pipeline/Pipeline.php
index de7a75dd..5a9775e8 100644
--- a/src/Pipeline/Pipeline.php
+++ b/src/Pipeline/Pipeline.php
@@ -4,7 +4,6 @@ declare(strict_types=1);
 
 namespace Patchlevel\EventSourcing\Pipeline;
 
-use Patchlevel\EventSourcing\Pipeline\Middleware\ChainMiddleware;
 use Patchlevel\EventSourcing\Pipeline\Middleware\Middleware;
 use Patchlevel\EventSourcing\Pipeline\Source\Source;
 use Patchlevel\EventSourcing\Pipeline\Target\Target;
@@ -13,7 +12,8 @@ class Pipeline
 {
     private Source $source;
     private Target $target;
-    private ChainMiddleware $middlewares;
+    /** @var list<Middleware> */
+    private array $middlewares;
 
     /**
      * @param list<Middleware> $middlewares
@@ -22,7 +22,7 @@ class Pipeline
     {
         $this->source = $source;
         $this->target = $target;
-        $this->middlewares = new ChainMiddleware($middlewares);
+        $this->middlewares = $middlewares;
     }
 
     /**
@@ -36,9 +36,7 @@ class Pipeline
         }
 
         foreach ($this->source->load() as $bucket) {
-            $result = ($this->middlewares)($bucket);
-
-            foreach ($result as $resultBucket) {
+            foreach ($this->processMiddlewares($bucket) as $resultBucket) {
                 $this->target->save($resultBucket);
             }
 
@@ -50,4 +48,34 @@ class Pipeline
     {
         return $this->source->count();
     }
+
+    /**
+     * @return list<EventBucket>
+     */
+    private function processMiddlewares(EventBucket $bucket): array
+    {
+        $buckets = [$bucket];
+
+        foreach ($this->middlewares as $middleware) {
+            $buckets = $this->processMiddleware($middleware, $buckets);
+        }
+
+        return $buckets;
+    }
+
+    /**
+     * @param list<EventBucket> $buckets
+     *
+     * @return list<EventBucket>
+     */
+    private function processMiddleware(Middleware $middleware, array $buckets): array
+    {
+        $result = [];
+
+        foreach ($buckets as $bucket) {
+            $result += $middleware($bucket);
+        }
+
+        return $result;
+    }
 }
diff --git a/src/Store/CorruptedMetadata.php b/src/Store/CorruptedMetadata.php
new file mode 100644
index 00000000..31430463
--- /dev/null
+++ b/src/Store/CorruptedMetadata.php
@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Store;
+
+use function sprintf;
+
+class CorruptedMetadata extends StoreException
+{
+    public function __construct(
+        string $expectedAggregateId,
+        string $expectedPlayhead,
+        string $actualAggregateId,
+        string $actualPlayhead
+    ) {
+        parent::__construct(sprintf(
+            'Corrupted metadata: %s:%s get %s:%s',
+            $actualAggregateId,
+            $actualPlayhead,
+            $expectedAggregateId,
+            $expectedPlayhead
+        ));
+    }
+}
diff --git a/src/Store/MultiTableStore.php b/src/Store/MultiTableStore.php
index 656666e0..09531c50 100644
--- a/src/Store/MultiTableStore.php
+++ b/src/Store/MultiTableStore.php
@@ -7,25 +7,33 @@ namespace Patchlevel\EventSourcing\Store;
 use Doctrine\DBAL\Connection;
 use Doctrine\DBAL\Schema\Schema;
 use Doctrine\DBAL\Types\Types;
+use Generator;
 use Patchlevel\EventSourcing\Aggregate\AggregateChanged;
 use Patchlevel\EventSourcing\Aggregate\AggregateRoot;
+use Patchlevel\EventSourcing\Pipeline\EventBucket;
 
+use function array_flip;
 use function array_key_exists;
 use function array_map;
 
-final class MultiTableStore extends DoctrineStore
+final class MultiTableStore extends DoctrineStore implements PipelineStore
 {
     /** @var array<class-string<AggregateRoot>, string> */
     private array $aggregates;
+    private string $metadataTableName;
 
     /**
      * @param array<class-string<AggregateRoot>, string> $aggregates
      */
-    public function __construct(Connection $eventConnection, array $aggregates)
-    {
+    public function __construct(
+        Connection $eventConnection,
+        array $aggregates,
+        string $metadataTableName = 'eventstore_metadata'
+    ) {
         parent::__construct($eventConnection);
 
         $this->aggregates = $aggregates;
+        $this->metadataTableName = $metadataTableName;
     }
 
     /**
@@ -95,38 +103,157 @@ final class MultiTableStore extends DoctrineStore
         $tableName = $this->tableName($aggregate);
 
         $this->connection->transactional(
-            static function (Connection $connection) use ($tableName, $id, $events): void {
+            function (Connection $connection) use ($tableName, $id, $events): void {
                 foreach ($events as $event) {
                     if ($event->aggregateId() !== $id) {
                         throw new StoreException('id missmatch');
                     }
 
-                    $data = $event->serialize();
-
-                    $connection->insert(
+                    $this->saveEvent(
+                        $connection,
                         $tableName,
-                        $data,
-                        [
-                            'recordedOn' => Types::DATETIMETZ_IMMUTABLE,
-                        ]
+                        $event
                     );
                 }
             }
         );
     }
 
+    public function all(): Generator
+    {
+        $queries = [];
+
+        foreach ($this->aggregates as $aggregate) {
+            $sql = $this->connection->createQueryBuilder()
+                ->select('*')
+                ->from($aggregate)
+                ->orderBy('id')
+                ->getSQL();
+
+            $query = $this->connection->iterateAssociative($sql);
+
+            if (!$query instanceof Generator) {
+                throw new StoreException();
+            }
+
+            $queries[$aggregate] = $query;
+        }
+
+        $sql = $this->connection->createQueryBuilder()
+            ->select('*')
+            ->from($this->metadataTableName)
+            ->orderBy('id')
+            ->getSQL();
+
+        $metaQuery = $this->connection->iterateAssociative($sql);
+
+        $platform = $this->connection->getDatabasePlatform();
+        $classMap = array_flip($this->aggregates);
+
+        foreach ($metaQuery as $metaData) {
+            $name = (string)$metaData['aggregate'];
+
+            if (!array_key_exists($name, $classMap)) {
+                throw new StoreException();
+            }
+
+            $eventData = $queries[$name]->current();
+            $queries[$name]->next();
+
+            if (
+                $eventData['aggregateId'] !== $metaData['aggregateId']
+                || $eventData['playhead'] !== $metaData['playhead']
+            ) {
+                throw new CorruptedMetadata(
+                    (string)$metaData['aggregateId'],
+                    (string)$metaData['playhead'],
+                    (string)$eventData['aggregateId'],
+                    (string)$eventData['playhead']
+                );
+            }
+
+            yield new EventBucket(
+                $classMap[$name],
+                AggregateChanged::deserialize(
+                    self::normalizeResult($platform, $eventData)
+                )
+            );
+        }
+    }
+
+    public function count(): int
+    {
+        $sql = $this->connection->createQueryBuilder()
+            ->select('COUNT(*)')
+            ->from($this->metadataTableName)
+            ->getSQL();
+
+        return (int)$this->connection->fetchOne($sql);
+    }
+
+    public function saveEventBucket(EventBucket $bucket): void
+    {
+        $this->saveEvent(
+            $this->connection,
+            $this->tableName($bucket->aggregateClass()),
+            $bucket->event()
+        );
+    }
+
+    private function saveEvent(Connection $connection, string $aggregate, AggregateChanged $event): void
+    {
+        $data = $event->serialize();
+
+        $connection->insert(
+            $aggregate,
+            $data,
+            [
+                'recordedOn' => Types::DATETIMETZ_IMMUTABLE,
+            ]
+        );
+
+        $connection->insert(
+            $this->metadataTableName,
+            [
+                'aggregate' => $aggregate,
+                'aggregateId' => $data['aggregateId'],
+                'playhead' => $data['playhead'],
+            ],
+        );
+    }
+
     public function schema(): Schema
     {
         $schema = new Schema([], [], $this->connection->getSchemaManager()->createSchemaConfig());
 
+        $this->addMetaTableToSchema($schema);
+
         foreach ($this->aggregates as $tableName) {
-            $this->addTableToSchema($schema, $tableName);
+            $this->addAggregateTableToSchema($schema, $tableName);
         }
 
         return $schema;
     }
 
-    private function addTableToSchema(Schema $schema, string $tableName): void
+    private function addMetaTableToSchema(Schema $schema): void
+    {
+        $table = $schema->createTable($this->metadataTableName);
+
+        $table->addColumn('id', Types::BIGINT)
+            ->setAutoincrement(true)
+            ->setNotnull(true);
+        $table->addColumn('aggregate', Types::STRING)
+            ->setNotnull(true);
+        $table->addColumn('aggregateId', Types::STRING)
+            ->setNotnull(true);
+        $table->addColumn('playhead', Types::INTEGER)
+            ->setNotnull(true);
+
+        $table->setPrimaryKey(['id']);
+        $table->addUniqueIndex(['aggregate', 'aggregateId', 'playhead']);
+    }
+
+    private function addAggregateTableToSchema(Schema $schema, string $tableName): void
     {
         $table = $schema->createTable($tableName);
 
diff --git a/src/Store/SingleTableStore.php b/src/Store/SingleTableStore.php
index bc3fe677..0008f2b6 100644
--- a/src/Store/SingleTableStore.php
+++ b/src/Store/SingleTableStore.php
@@ -20,17 +20,17 @@ final class SingleTableStore extends DoctrineStore implements PipelineStore
 {
     /** @var array<class-string<AggregateRoot>, string> */
     private array $aggregates;
-    private string $tableName;
+    private string $storeTableName;
 
     /**
      * @param array<class-string<AggregateRoot>, string> $aggregates
      */
-    public function __construct(Connection $connection, array $aggregates, string $tableName)
+    public function __construct(Connection $connection, array $aggregates, string $storeTableName)
     {
         parent::__construct($connection);
 
         $this->aggregates = $aggregates;
-        $this->tableName = $tableName;
+        $this->storeTableName = $storeTableName;
     }
 
     /**
@@ -44,7 +44,7 @@ final class SingleTableStore extends DoctrineStore implements PipelineStore
 
         $sql = $this->connection->createQueryBuilder()
             ->select('*')
-            ->from($this->tableName)
+            ->from($this->storeTableName)
             ->where('aggregate = :aggregate AND aggregateId = :id AND playhead > :playhead')
             ->getSQL();
 
@@ -79,7 +79,7 @@ final class SingleTableStore extends DoctrineStore implements PipelineStore
 
         $sql = $this->connection->createQueryBuilder()
             ->select('COUNT(*)')
-            ->from($this->tableName)
+            ->from($this->storeTableName)
             ->where('aggregate = :aggregate AND aggregateId = :id')
             ->setMaxResults(1)
             ->getSQL();
@@ -102,10 +102,10 @@ final class SingleTableStore extends DoctrineStore implements PipelineStore
     public function saveBatch(string $aggregate, string $id, array $events): void
     {
         $shortName = $this->shortName($aggregate);
-        $tableName = $this->tableName;
+        $storeTableName = $this->storeTableName;
 
         $this->connection->transactional(
-            static function (Connection $connection) use ($shortName, $id, $events, $tableName): void {
+            static function (Connection $connection) use ($shortName, $id, $events, $storeTableName): void {
                 foreach ($events as $event) {
                     if ($event->aggregateId() !== $id) {
                         throw new StoreException('id missmatch');
@@ -115,7 +115,7 @@ final class SingleTableStore extends DoctrineStore implements PipelineStore
                     $data['aggregate'] = $shortName;
 
                     $connection->insert(
-                        $tableName,
+                        $storeTableName,
                         $data,
                         [
                             'recordedOn' => Types::DATETIMETZ_IMMUTABLE,
@@ -133,7 +133,7 @@ final class SingleTableStore extends DoctrineStore implements PipelineStore
     {
         $sql = $this->connection->createQueryBuilder()
             ->select('*')
-            ->from($this->tableName)
+            ->from($this->storeTableName)
             ->orderBy('id')
             ->getSQL();
 
@@ -164,7 +164,7 @@ final class SingleTableStore extends DoctrineStore implements PipelineStore
     {
         $sql = $this->connection->createQueryBuilder()
             ->select('COUNT(*)')
-            ->from($this->tableName)
+            ->from($this->storeTableName)
             ->getSQL();
 
         return (int)$this->connection->fetchOne($sql);
@@ -176,7 +176,7 @@ final class SingleTableStore extends DoctrineStore implements PipelineStore
         $data['aggregate'] = $this->shortName($bucket->aggregateClass());
 
         $this->connection->insert(
-            $this->tableName,
+            $this->storeTableName,
             $data,
             [
                 'recordedOn' => Types::DATETIMETZ_IMMUTABLE,
@@ -187,14 +187,7 @@ final class SingleTableStore extends DoctrineStore implements PipelineStore
     public function schema(): Schema
     {
         $schema = new Schema([], [], $this->connection->getSchemaManager()->createSchemaConfig());
-        $this->addTableToSchema($schema);
-
-        return $schema;
-    }
-
-    private function addTableToSchema(Schema $schema): void
-    {
-        $table = $schema->createTable($this->tableName);
+        $table = $schema->createTable($this->storeTableName);
 
         $table->addColumn('id', Types::BIGINT)
             ->setAutoincrement(true)
@@ -213,8 +206,9 @@ final class SingleTableStore extends DoctrineStore implements PipelineStore
             ->setNotnull(false);
 
         $table->setPrimaryKey(['id']);
-
         $table->addUniqueIndex(['aggregate', 'aggregateId', 'playhead']);
+
+        return $schema;
     }
 
     /**


Test patch:
diff --git a/tests/Integration/Pipeline/PipelineChangeStoreTest.php b/tests/Integration/Pipeline/PipelineChangeStoreTest.php
index 2e3afe12..11da528f 100644
--- a/tests/Integration/Pipeline/PipelineChangeStoreTest.php
+++ b/tests/Integration/Pipeline/PipelineChangeStoreTest.php
@@ -8,7 +8,7 @@ use Doctrine\DBAL\Connection;
 use Doctrine\DBAL\Driver\PDO\SQLite\Driver;
 use Doctrine\DBAL\DriverManager;
 use Patchlevel\EventSourcing\EventBus\DefaultEventBus;
-use Patchlevel\EventSourcing\Pipeline\Middleware\ExcludeEventMiddleware;
+use Patchlevel\EventSourcing\Pipeline\Middleware\DeleteEventMiddleware;
 use Patchlevel\EventSourcing\Pipeline\Middleware\RecalculatePlayheadMiddleware;
 use Patchlevel\EventSourcing\Pipeline\Middleware\ReplaceEventMiddleware;
 use Patchlevel\EventSourcing\Pipeline\Pipeline;
@@ -16,6 +16,7 @@ use Patchlevel\EventSourcing\Pipeline\Source\StoreSource;
 use Patchlevel\EventSourcing\Pipeline\Target\StoreTarget;
 use Patchlevel\EventSourcing\Repository\Repository;
 use Patchlevel\EventSourcing\Schema\DoctrineSchemaManager;
+use Patchlevel\EventSourcing\Store\MultiTableStore;
 use Patchlevel\EventSourcing\Store\SingleTableStore;
 use Patchlevel\EventSourcing\Tests\Integration\Pipeline\Aggregate\Profile;
 use Patchlevel\EventSourcing\Tests\Integration\Pipeline\Events\NewVisited;
@@ -66,7 +67,7 @@ final class PipelineChangeStoreTest extends TestCase
 
     public function testSuccessful(): void
     {
-        $oldStore = new SingleTableStore(
+        $oldStore = new MultiTableStore(
             $this->connectionOld,
             [Profile::class => 'profile'],
             'eventstore'
@@ -101,7 +102,7 @@ final class PipelineChangeStoreTest extends TestCase
             new StoreSource($oldStore),
             new StoreTarget($newStore),
             [
-                new ExcludeEventMiddleware([PrivacyAdded::class]),
+                new DeleteEventMiddleware([PrivacyAdded::class]),
                 new ReplaceEventMiddleware(OldVisited::class, static function (OldVisited $oldVisited) {
                     return NewVisited::raise($oldVisited->profileId());
                 }),
diff --git a/tests/Unit/Fixture/AliasProfileCreated.php b/tests/Unit/Fixture/AliasProfileCreated.php
deleted file mode 100644
index 7c3434e8..00000000
--- a/tests/Unit/Fixture/AliasProfileCreated.php
+++ /dev/null
@@ -1,10 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Patchlevel\EventSourcing\Tests\Unit\Fixture;
-
-final class AliasProfileCreated extends ProfileCreated
-{
-    // .. alias, simulate class rename/movement
-}
diff --git a/tests/Unit/Fixture/ProfileCreated.php b/tests/Unit/Fixture/ProfileCreated.php
index 98037947..c19130f7 100644
--- a/tests/Unit/Fixture/ProfileCreated.php
+++ b/tests/Unit/Fixture/ProfileCreated.php
@@ -6,7 +6,7 @@ namespace Patchlevel\EventSourcing\Tests\Unit\Fixture;
 
 use Patchlevel\EventSourcing\Aggregate\AggregateChanged;
 
-class ProfileCreated extends AggregateChanged
+final class ProfileCreated extends AggregateChanged
 {
     public static function raise(
         ProfileId $id,
diff --git a/tests/Unit/Pipeline/Middleware/ClassRenameMiddlewareTest.php b/tests/Unit/Pipeline/Middleware/ClassRenameMiddlewareTest.php
deleted file mode 100644
index 1144bd90..00000000
--- a/tests/Unit/Pipeline/Middleware/ClassRenameMiddlewareTest.php
+++ /dev/null
@@ -1,68 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Patchlevel\EventSourcing\Tests\Unit\Pipeline;
-
-use Patchlevel\EventSourcing\Pipeline\EventBucket;
-use Patchlevel\EventSourcing\Pipeline\Middleware\ClassRenameMiddleware;
-use Patchlevel\EventSourcing\Tests\Unit\Fixture\AliasProfileCreated;
-use Patchlevel\EventSourcing\Tests\Unit\Fixture\Email;
-use Patchlevel\EventSourcing\Tests\Unit\Fixture\MessagePublished;
-use Patchlevel\EventSourcing\Tests\Unit\Fixture\Profile;
-use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileCreated;
-use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileId;
-use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileVisited;
-use PHPUnit\Framework\TestCase;
-
-class ClassRenameMiddlewareTest extends TestCase
-{
-    public function testRename(): void
-    {
-        $middleware = new ClassRenameMiddleware([
-            AliasProfileCreated::class => ProfileCreated::class,
-        ]);
-
-        $event = AliasProfileCreated::raise(
-            ProfileId::fromString('1'),
-            Email::fromString('d.a.badura@gmail.com')
-        )->recordNow(5);
-
-        $bucket = new EventBucket(
-            Profile::class,
-            $event
-        );
-
-        $result = $middleware($bucket);
-
-        self::assertCount(1, $result);
-
-        $newEvent = $result[0]->event();
-
-        self::assertInstanceOf(ProfileCreated::class, $newEvent);
-        self::assertNotInstanceOf(AliasProfileCreated::class, $newEvent);
-        self::assertEquals($event->payload(), $newEvent->payload());
-        self::assertEquals($event->playhead(), $newEvent->playhead());
-        self::assertEquals($event->recordedOn(), $newEvent->recordedOn());
-        self::assertEquals($event->aggregateId(), $newEvent->aggregateId());
-    }
-
-    public function testSkip(): void
-    {
-        $middleware = new ClassRenameMiddleware([
-            ProfileVisited::class => MessagePublished::class,
-        ]);
-
-        $bucket = new EventBucket(
-            Profile::class,
-            AliasProfileCreated::raise(
-                ProfileId::fromString('1'),
-                Email::fromString('d.a.badura@gmail.com')
-            )->recordNow(5)
-        );
-
-        $result = $middleware($bucket);
-
-        self::assertEquals([$bucket], $result);
-    }
-}
diff --git a/tests/Unit/Pipeline/Middleware/ExcludeEventMiddlewareTest.php b/tests/Unit/Pipeline/Middleware/DeleteEventMiddlewareTest.php
similarity index 81%
rename from tests/Unit/Pipeline/Middleware/ExcludeEventMiddlewareTest.php
rename to tests/Unit/Pipeline/Middleware/DeleteEventMiddlewareTest.php
index 01e3e507..a7399182 100644
--- a/tests/Unit/Pipeline/Middleware/ExcludeEventMiddlewareTest.php
+++ b/tests/Unit/Pipeline/Middleware/DeleteEventMiddlewareTest.php
@@ -5,7 +5,7 @@ declare(strict_types=1);
 namespace Patchlevel\EventSourcing\Tests\Unit\Pipeline;
 
 use Patchlevel\EventSourcing\Pipeline\EventBucket;
-use Patchlevel\EventSourcing\Pipeline\Middleware\ExcludeEventMiddleware;
+use Patchlevel\EventSourcing\Pipeline\Middleware\DeleteEventMiddleware;
 use Patchlevel\EventSourcing\Tests\Unit\Fixture\Email;
 use Patchlevel\EventSourcing\Tests\Unit\Fixture\Profile;
 use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileCreated;
@@ -13,11 +13,11 @@ use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileId;
 use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileVisited;
 use PHPUnit\Framework\TestCase;
 
-class ExcludeEventMiddlewareTest extends TestCase
+class DeleteEventMiddlewareTest extends TestCase
 {
     public function testDeleteEvent(): void
     {
-        $middleware = new ExcludeEventMiddleware([ProfileCreated::class]);
+        $middleware = new DeleteEventMiddleware([ProfileCreated::class]);
 
         $bucket = new EventBucket(
             Profile::class,
@@ -34,7 +34,7 @@ class ExcludeEventMiddlewareTest extends TestCase
 
     public function testSkipEvent(): void
     {
-        $middleware = new ExcludeEventMiddleware([ProfileCreated::class]);
+        $middleware = new DeleteEventMiddleware([ProfileCreated::class]);
 
         $bucket = new EventBucket(
             Profile::class,
diff --git a/tests/Unit/Pipeline/Middleware/FilterEventMiddlewareTest.php b/tests/Unit/Pipeline/Middleware/FilterEventMiddlewareTest.php
index 8dd05a17..23a1f8c1 100644
--- a/tests/Unit/Pipeline/Middleware/FilterEventMiddlewareTest.php
+++ b/tests/Unit/Pipeline/Middleware/FilterEventMiddlewareTest.php
@@ -4,7 +4,6 @@ declare(strict_types=1);
 
 namespace Patchlevel\EventSourcing\Tests\Unit\Pipeline;
 
-use Patchlevel\EventSourcing\Aggregate\AggregateChanged;
 use Patchlevel\EventSourcing\Pipeline\EventBucket;
 use Patchlevel\EventSourcing\Pipeline\Middleware\FilterEventMiddleware;
 use Patchlevel\EventSourcing\Tests\Unit\Fixture\Email;
@@ -16,11 +15,9 @@ use PHPUnit\Framework\TestCase;
 
 class FilterEventMiddlewareTest extends TestCase
 {
-    public function testPositive(): void
+    public function testFilterEvent(): void
     {
-        $middleware = new FilterEventMiddleware(static function (AggregateChanged $aggregateChanged) {
-            return $aggregateChanged instanceof ProfileCreated;
-        });
+        $middleware = new FilterEventMiddleware([ProfileCreated::class]);
 
         $bucket = new EventBucket(
             Profile::class,
@@ -35,11 +32,9 @@ class FilterEventMiddlewareTest extends TestCase
         self::assertEquals([$bucket], $result);
     }
 
-    public function testNegative(): void
+    public function testSkipEvent(): void
     {
-        $middleware = new FilterEventMiddleware(static function (AggregateChanged $aggregateChanged) {
-            return $aggregateChanged instanceof ProfileCreated;
-        });
+        $middleware = new FilterEventMiddleware([ProfileCreated::class]);
 
         $bucket = new EventBucket(
             Profile::class,
diff --git a/tests/Unit/Pipeline/Middleware/IncludeEventMiddlewareTest.php b/tests/Unit/Pipeline/Middleware/IncludeEventMiddlewareTest.php
deleted file mode 100644
index c8d495cc..00000000
--- a/tests/Unit/Pipeline/Middleware/IncludeEventMiddlewareTest.php
+++ /dev/null
@@ -1,51 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Patchlevel\EventSourcing\Tests\Unit\Pipeline;
-
-use Patchlevel\EventSourcing\Pipeline\EventBucket;
-use Patchlevel\EventSourcing\Pipeline\Middleware\IncludeEventMiddleware;
-use Patchlevel\EventSourcing\Tests\Unit\Fixture\Email;
-use Patchlevel\EventSourcing\Tests\Unit\Fixture\Profile;
-use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileCreated;
-use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileId;
-use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileVisited;
-use PHPUnit\Framework\TestCase;
-
-class IncludeEventMiddlewareTest extends TestCase
-{
-    public function testFilterEvent(): void
-    {
-        $middleware = new IncludeEventMiddleware([ProfileCreated::class]);
-
-        $bucket = new EventBucket(
-            Profile::class,
-            ProfileCreated::raise(
-                ProfileId::fromString('1'),
-                Email::fromString('d.a.badura@gmail.com')
-            )->recordNow(0)
-        );
-
-        $result = $middleware($bucket);
-
-        self::assertEquals([$bucket], $result);
-    }
-
-    public function testSkipEvent(): void
-    {
-        $middleware = new IncludeEventMiddleware([ProfileCreated::class]);
-
-        $bucket = new EventBucket(
-            Profile::class,
-            ProfileVisited::raise(
-                ProfileId::fromString('1'),
-                ProfileId::fromString('2')
-            )->recordNow(0)
-        );
-
-        $result = $middleware($bucket);
-
-        self::assertEquals([], $result);
-    }
-}
diff --git a/tests/Unit/Pipeline/PipelineTest.php b/tests/Unit/Pipeline/PipelineTest.php
index 3402d3e1..4f39b1e1 100644
--- a/tests/Unit/Pipeline/PipelineTest.php
+++ b/tests/Unit/Pipeline/PipelineTest.php
@@ -5,7 +5,7 @@ declare(strict_types=1);
 namespace Patchlevel\EventSourcing\Tests\Unit\Pipeline;
 
 use Patchlevel\EventSourcing\Pipeline\EventBucket;
-use Patchlevel\EventSourcing\Pipeline\Middleware\ExcludeEventMiddleware;
+use Patchlevel\EventSourcing\Pipeline\Middleware\DeleteEventMiddleware;
 use Patchlevel\EventSourcing\Pipeline\Middleware\RecalculatePlayheadMiddleware;
 use Patchlevel\EventSourcing\Pipeline\Pipeline;
 use Patchlevel\EventSourcing\Pipeline\Source\InMemorySource;
@@ -116,7 +116,7 @@ class PipelineTest extends TestCase
             $source,
             $target,
             [
-                new ExcludeEventMiddleware([ProfileCreated::class]),
+                new DeleteEventMiddleware([ProfileCreated::class]),
                 new RecalculatePlayheadMiddleware(),
             ]
         );