filter events in subscription engine

deptrac.yaml

@@ -154,6 +154,7 @@ deptrac:
       - Attribute
       - Clock
       - Message
+      - MetadataEvent
       - MetadataSubscriber
       - Repository
       - Schema

src/Store/Criteria/EventsCriterion.php

@@ -0,0 +1,14 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Store\Criteria;
+
+final class EventsCriterion
+{
+    public function __construct(
+        /** @var list<string> */
+        public readonly array $events,
+    ) {
+    }
+}

src/Store/DoctrineDbalStore.php

@@ -5,6 +5,7 @@
 namespace Patchlevel\EventSourcing\Store;
 
 use Closure;
+use Doctrine\DBAL\ArrayParameterType;
 use Doctrine\DBAL\Connection;
 use Doctrine\DBAL\Exception\UniqueConstraintViolationException;
 use Doctrine\DBAL\Platforms\MariaDBPlatform;
@@ -26,6 +27,7 @@
 use Patchlevel\EventSourcing\Store\Criteria\AggregateNameCriterion;
 use Patchlevel\EventSourcing\Store\Criteria\ArchivedCriterion;
 use Patchlevel\EventSourcing\Store\Criteria\Criteria;
+use Patchlevel\EventSourcing\Store\Criteria\EventsCriterion;
 use Patchlevel\EventSourcing\Store\Criteria\FromIndexCriterion;
 use Patchlevel\EventSourcing\Store\Criteria\FromPlayheadCriterion;
 use PDO;
@@ -156,6 +158,10 @@ private function applyCriteria(QueryBuilder $builder, Criteria $criteria): void
                     $builder->andWhere('id > :index');
                     $builder->setParameter('index', $criterion->fromIndex, Types::INTEGER);
                     break;
+                case EventsCriterion::class:
+                    $builder->andWhere('event IN (:events)');
+                    $builder->setParameter('events', $criterion->events, ArrayParameterType::STRING);
+                    break;
                 default:
                     throw new UnsupportedCriterion($criterion::class);
             }

src/Store/StreamDoctrineDbalStore.php

@@ -5,6 +5,7 @@
 namespace Patchlevel\EventSourcing\Store;
 
 use Closure;
+use Doctrine\DBAL\ArrayParameterType;
 use Doctrine\DBAL\Connection;
 use Doctrine\DBAL\Exception\UniqueConstraintViolationException;
 use Doctrine\DBAL\Platforms\MariaDBPlatform;
@@ -24,6 +25,7 @@
 use Patchlevel\EventSourcing\Serializer\EventSerializer;
 use Patchlevel\EventSourcing\Store\Criteria\ArchivedCriterion;
 use Patchlevel\EventSourcing\Store\Criteria\Criteria;
+use Patchlevel\EventSourcing\Store\Criteria\EventsCriterion;
 use Patchlevel\EventSourcing\Store\Criteria\FromIndexCriterion;
 use Patchlevel\EventSourcing\Store\Criteria\FromPlayheadCriterion;
 use Patchlevel\EventSourcing\Store\Criteria\StreamCriterion;
@@ -169,6 +171,10 @@ private function applyCriteria(QueryBuilder $builder, Criteria $criteria): void
                     $builder->andWhere('id > :index');
                     $builder->setParameter('index', $criterion->fromIndex, Types::INTEGER);
                     break;
+                case EventsCriterion::class:
+                    $builder->andWhere('event IN (:events)');
+                    $builder->setParameter('events', $criterion->events, ArrayParameterType::STRING);
+                    break;
                 default:
                     throw new UnsupportedCriterion($criterion::class);
             }

src/Subscription/Engine/DefaultSubscriptionEngine.php

@@ -5,8 +5,6 @@
 namespace Patchlevel\EventSourcing\Subscription\Engine;
 
 use Patchlevel\EventSourcing\Message\Message;
-use Patchlevel\EventSourcing\Store\Criteria\Criteria;
-use Patchlevel\EventSourcing\Store\Criteria\FromIndexCriterion;
 use Patchlevel\EventSourcing\Store\Store;
 use Patchlevel\EventSourcing\Subscription\RetryStrategy\ClockBasedRetryStrategy;
 use Patchlevel\EventSourcing\Subscription\RetryStrategy\RetryStrategy;
@@ -35,13 +33,21 @@ final class DefaultSubscriptionEngine implements SubscriptionEngine
     /** @var array<string, BatchableSubscriber> */
     private array $batching = [];
 
+    private readonly MessageLoader $messageLoader;
+
     public function __construct(
-        private readonly Store $messageStore,
+        Store|MessageLoader $messageStore,
         SubscriptionStore $subscriptionStore,
         private readonly SubscriberAccessorRepository $subscriberRepository,
         private readonly RetryStrategy $retryStrategy = new ClockBasedRetryStrategy(),
         private readonly LoggerInterface|null $logger = null,
     ) {
+        if ($messageStore instanceof MessageLoader) {
+            $this->messageLoader = $messageStore;
+        } else {
+            $this->messageLoader = new StoreMessageLoader($messageStore);
+        }
+
         $this->subscriptionManager = new SubscriptionManager($subscriptionStore);
     }
 
@@ -72,7 +78,7 @@ function (SubscriptionCollection $subscriptions) use ($skipBooting): Result {
                 /** @var list<Error> $errors */
                 $errors = [];
 
-                $latestIndex = $this->latestIndex();
+                $latestIndex = $this->messageLoader->lastIndex();
 
                 foreach ($subscriptions as $subscription) {
                     $subscriber = $this->subscriber($subscription->id());
@@ -193,9 +199,7 @@ function (SubscriptionCollection $subscriptions) use ($limit): ProcessedResult {
                     $messageCounter = 0;
 
                     try {
-                        $stream = $this->messageStore->load(
-                            new Criteria(new FromIndexCriterion($startIndex)),
-                        );
+                        $stream = $this->messageLoader->load($startIndex, $subscriptions->toArray());
 
                         foreach ($stream as $message) {
                             $messageCounter++;
@@ -366,8 +370,7 @@ function (SubscriptionCollection $subscriptions) use ($limit): ProcessedResult {
                     $messageCounter = 0;
 
                     try {
-                        $criteria = new Criteria(new FromIndexCriterion($startIndex));
-                        $stream = $this->messageStore->load($criteria);
+                        $stream = $this->messageLoader->load($startIndex, $subscriptions->toArray());
 
                         foreach ($stream as $message) {
                             $messageCounter++;
@@ -966,7 +969,7 @@ private function discoverNewSubscriptions(): void
 
             if ($subscriber->setupMethod() === null && $subscriber->runMode() === RunMode::FromNow) {
                 if ($latestIndex === null) {
-                    $latestIndex = $this->latestIndex();
+                    $latestIndex = $this->messageLoader->lastIndex();
                 }
 
                 $subscription->changePosition($latestIndex);
@@ -986,13 +989,6 @@ private function discoverNewSubscriptions(): void
         $this->subscriptionManager->flush();
     }
 
-    private function latestIndex(): int
-    {
-        $stream = $this->messageStore->load(null, 1, null, true);
-
-        return $stream->index() ?: 0;
-    }
-
     private function handleError(Subscription $subscription, Throwable $throwable): void
     {
         $subscription->error($throwable);

src/Subscription/Engine/EventFilteredStoreMessageLoader.php

@@ -0,0 +1,84 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine;
+
+use Patchlevel\EventSourcing\Metadata\Event\EventMetadataFactory;
+use Patchlevel\EventSourcing\Store\Criteria\Criteria;
+use Patchlevel\EventSourcing\Store\Criteria\EventsCriterion;
+use Patchlevel\EventSourcing\Store\Criteria\FromIndexCriterion;
+use Patchlevel\EventSourcing\Store\Store;
+use Patchlevel\EventSourcing\Store\Stream;
+use Patchlevel\EventSourcing\Subscription\Subscriber\MetadataSubscriberAccessor;
+use Patchlevel\EventSourcing\Subscription\Subscriber\SubscriberAccessorRepository;
+use Patchlevel\EventSourcing\Subscription\Subscription;
+
+use function array_keys;
+
+final class EventFilteredStoreMessageLoader implements MessageLoader
+{
+    public function __construct(
+        private readonly Store $store,
+        private readonly EventMetadataFactory $eventMetadataFactory,
+        private readonly SubscriberAccessorRepository $subscriberRepository,
+    ) {
+    }
+
+    /** @param list<Subscription> $subscriptions */
+    public function load(int $startIndex, array $subscriptions): Stream
+    {
+        $criteria = new Criteria(new FromIndexCriterion($startIndex));
+
+        $events = $this->events($subscriptions);
+
+        if ($events !== []) {
+            $criteria = $criteria->add(new EventsCriterion($events));
+        }
+
+        return $this->store->load($criteria);
+    }
+
+    /**
+     * @param list<Subscription> $subscriptions
+     *
+     * @return list<string>
+     */
+    private function events(array $subscriptions): array
+    {
+        $eventNames = [];
+
+        foreach ($subscriptions as $subscription) {
+            $subscriber =  $this->subscriberRepository->get($subscription->id());
+
+            if (!$subscriber instanceof MetadataSubscriberAccessor) {
+                return [];
+            }
+
+            $events = $subscriber->events();
+
+            foreach ($events as $event) {
+                if ($event === '*') {
+                    return [];
+                }
+
+                $metadata = $this->eventMetadataFactory->metadata($event);
+
+                $eventNames[$metadata->name] = true;
+
+                foreach ($metadata->aliases as $alias) {
+                    $eventNames[$alias] = true;
+                }
+            }
+        }
+
+        return array_keys($eventNames);
+    }
+
+    public function lastIndex(): int
+    {
+        $stream = $this->store->load(null, 1, null, true);
+
+        return $stream->index() ?: 0;
+    }
+}

src/Subscription/Engine/MessageLoader.php

@@ -0,0 +1,16 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine;
+
+use Patchlevel\EventSourcing\Store\Stream;
+use Patchlevel\EventSourcing\Subscription\Subscription;
+
+interface MessageLoader
+{
+    /** @param list<Subscription> $subscriptions */
+    public function load(int $startIndex, array $subscriptions): Stream;
+
+    public function lastIndex(): int;
+}

src/Subscription/Engine/StoreMessageLoader.php

@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Engine;
+
+use Patchlevel\EventSourcing\Store\Criteria\Criteria;
+use Patchlevel\EventSourcing\Store\Criteria\FromIndexCriterion;
+use Patchlevel\EventSourcing\Store\Store;
+use Patchlevel\EventSourcing\Store\Stream;
+use Patchlevel\EventSourcing\Subscription\Subscription;
+
+final class StoreMessageLoader implements MessageLoader
+{
+    public function __construct(
+        private readonly Store $store,
+    ) {
+    }
+
+    /** @param list<Subscription> $subscriptions */
+    public function load(int $startIndex, array $subscriptions): Stream
+    {
+        return $this->store->load(new Criteria(new FromIndexCriterion($startIndex)));
+    }
+
+    public function lastIndex(): int
+    {
+        $stream = $this->store->load(null, 1, null, true);
+
+        return $stream->index() ?: 0;
+    }
+}

src/Subscription/Engine/SubscriptionCollection.php

@@ -64,4 +64,10 @@ public function lowestPosition(): int
 
         return $min;
     }
+
+    /** @return list<Subscription> */
+    public function toArray(): array
+    {
+        return $this->subscriptions;
+    }
 }

src/Subscription/Subscriber/MetadataSubscriberAccessor.php

@@ -13,6 +13,7 @@
 use Patchlevel\EventSourcing\Subscription\Subscriber\ArgumentResolver\ArgumentResolver;
 
 use function array_key_exists;
+use function array_keys;
 use function array_map;
 use function array_merge;
 
@@ -84,6 +85,12 @@ public function teardownMethod(): Closure|null
         return $this->subscriber->$method(...);
     }
 
+    /** @return list<class-string|'*'> */
+    public function events(): array
+    {
+        return array_keys($this->metadata->subscribeMethods);
+    }
+
     /**
      * @param class-string $eventClass
      *

tests/Benchmark/SubscriptionEngineBench.php

@@ -5,6 +5,7 @@
 namespace Patchlevel\EventSourcing\Tests\Benchmark;
 
 use Patchlevel\EventSourcing\Aggregate\AggregateRootId;
+use Patchlevel\EventSourcing\Metadata\Event\AttributeEventMetadataFactory;
 use Patchlevel\EventSourcing\Repository\DefaultRepository;
 use Patchlevel\EventSourcing\Repository\Repository;
 use Patchlevel\EventSourcing\Schema\ChainDoctrineSchemaConfigurator;
@@ -13,6 +14,7 @@
 use Patchlevel\EventSourcing\Store\DoctrineDbalStore;
 use Patchlevel\EventSourcing\Store\Store;
 use Patchlevel\EventSourcing\Subscription\Engine\DefaultSubscriptionEngine;
+use Patchlevel\EventSourcing\Subscription\Engine\EventFilteredStoreMessageLoader;
 use Patchlevel\EventSourcing\Subscription\Engine\SubscriptionEngine;
 use Patchlevel\EventSourcing\Subscription\Store\DoctrineSubscriptionStore;
 use Patchlevel\EventSourcing\Subscription\Subscriber\MetadataSubscriberAccessorRepository;
@@ -63,20 +65,27 @@ public function setUp(): void
         $profile = Profile::create($this->id, 'Peter');
 
         for ($i = 1; $i < 10_000; $i++) {
+            $profile->changeEmail('peter' . $i . '@example.com');
             $profile->changeName('Peter ' . $i);
         }
 
         $this->repository->save($profile);
 
+        $subscriberAccessorRepository = new MetadataSubscriberAccessorRepository(
+            [
+                new ProfileProjector($connection),
+                new SendEmailProcessor(),
+            ],
+        );
+
         $this->subscriptionEngine = new DefaultSubscriptionEngine(
-            $this->store,
-            $subscriptionStore,
-            new MetadataSubscriberAccessorRepository(
-                [
-                    new ProfileProjector($connection),
-                    new SendEmailProcessor(),
-                ],
+            new EventFilteredStoreMessageLoader(
+                $this->store,
+                new AttributeEventMetadataFactory(),
+                $subscriberAccessorRepository,
             ),
+            $subscriptionStore,
+            $subscriberAccessorRepository,
         );
     }