Merge pull request #509 from suhailhijry/main

Event replay optimizations

Author: freekmurze

src/Console/ReplayCommand.php

@@ -6,6 +6,7 @@
 use Illuminate\Console\Command;
 use Illuminate\Support\Collection;
 use Spatie\EventSourcing\Projectionist;
+use Spatie\EventSourcing\EventHandlers\Projectors\Projector;
 use Spatie\EventSourcing\StoredEvents\Repositories\StoredEventRepository;
 
 class ReplayCommand extends Command
@@ -67,7 +68,8 @@ public function selectProjectors(array $projectorClassNames): ?Collection
     public function replay(Collection $projectors, int $startingFrom, ?string $aggregateUuid = null): void
     {
         $repository = app(StoredEventRepository::class);
-        $replayCount = $repository->countAllStartingFrom($startingFrom, $aggregateUuid);
+        $events = collect($projectors->toArray())->map(fn(Projector $projector) => $projector->getEventHandlingMethods()->keys())->flatten()->toArray();
+        $replayCount = $repository->countAllStartingFrom($startingFrom, $aggregateUuid, $events);
 
         if ($replayCount === 0) {
             $this->warn('There are no events to replay');

src/Projectionist.php

@@ -322,6 +322,7 @@ public function replay(
         ?callable $onEventReplayed = null,
         ?string $aggregateUuid = null
     ): void {
+        $events = collect($projectors->toArray())->map(fn(Projector $projector) => $projector->getEventHandlingMethods()->keys())->flatten()->toArray();
         $projectors = (new EventHandlerCollection($projectors))
             ->sortBy(fn (EventHandler $eventHandler) => $eventHandler->getWeight(null));
 
@@ -340,8 +341,7 @@ public function replay(
         $projectors->call('onStartingEventReplay');
 
         app(StoredEventRepository::class)
-            ->retrieveAllStartingFrom($startingFromEventId, $aggregateUuid)
-            ->each(function (StoredEvent $storedEvent) use ($projectors, $onEventReplayed) {
+            ->runForAllStartingFrom($startingFromEventId, function (StoredEvent $storedEvent) use ($projectors, $onEventReplayed) {
                 $this->applyStoredEventToProjectors(
                     $storedEvent,
                     $projectors->forEvent($storedEvent)
@@ -350,7 +350,7 @@ public function replay(
                 if ($onEventReplayed) {
                     $onEventReplayed($storedEvent);
                 }
-            });
+            }, 1000, $aggregateUuid, $events);
 
         $this->isReplaying = false;
 

src/StoredEvents/Repositories/EloquentStoredEventRepository.php

@@ -5,6 +5,7 @@
 use Carbon\Carbon;
 use Illuminate\Database\Eloquent\Builder;
 use Illuminate\Support\LazyCollection;
+use Illuminate\Support\Collection;
 use ReflectionClass;
 use ReflectionException;
 use Spatie\EventSourcing\AggregateRoots\Exceptions\InvalidEloquentStoredEventModel;
@@ -49,9 +50,9 @@ public function retrieveAll(?string $uuid = null): LazyCollection
         return $query->orderBy('id')->cursor()->map(fn (EloquentStoredEvent $storedEvent) => $storedEvent->toStoredEvent());
     }
 
-    public function retrieveAllStartingFrom(int $startingFrom, ?string $uuid = null): LazyCollection
+    public function retrieveAllStartingFrom(int $startingFrom, ?string $uuid = null, array $events = []): LazyCollection
     {
-        $query = $this->prepareEventModelQuery($startingFrom, $uuid);
+        $query = $this->prepareEventModelQuery($startingFrom, $uuid, $events);
 
         /** @var LazyCollection $lazyCollection */
         $lazyCollection = $query
@@ -61,9 +62,24 @@ public function retrieveAllStartingFrom(int $startingFrom, ?string $uuid = null)
         return $lazyCollection->map(fn (EloquentStoredEvent $storedEvent) => $storedEvent->toStoredEvent());
     }
 
-    public function countAllStartingFrom(int $startingFrom, ?string $uuid = null): int
+
+    public function runForAllStartingFrom(int $startingFrom, callable|\Closure $function, int $chunkSize = 1000, ?string $uuid = null, array $events = []): bool {
+        $query = $this->prepareEventModelQuery($startingFrom, $uuid, $events);
+
+        $query = $query
+            ->orderBy('id');
+
+        return $query->chunk($chunkSize, function (Collection $events) use ($function) {
+            foreach ($events as $event) {
+                $storedEVent = $event->toStoredEvent();
+                $function($storedEVent);
+            }
+        });
+    }
+
+    public function countAllStartingFrom(int $startingFrom, ?string $uuid = null, array $events = []): int
     {
-        return $this->prepareEventModelQuery($startingFrom, $uuid)->count('id');
+        return $this->prepareEventModelQuery($startingFrom, $uuid, $events)->count('id');
     }
 
     public function retrieveAllAfterVersion(int $aggregateVersion, string $aggregateUuid): LazyCollection
@@ -171,14 +187,18 @@ public function getLatestAggregateVersion(string $aggregateUuid): int
                 ->max('aggregate_version') ?? 0;
     }
 
-    private function prepareEventModelQuery(int $startingFrom, ?string $uuid = null): Builder
+    private function prepareEventModelQuery(int $startingFrom, ?string $uuid = null, array $events = []): Builder
     {
         $query = $this->getQuery()->startingFrom($startingFrom);
 
         if ($uuid) {
             $query->whereAggregateRoot($uuid);
         }
 
+        if (!empty($events)) {
+            $query->whereEvent(...$events);
+        }
+
         return $query;
     }
 

src/StoredEvents/Repositories/StoredEventRepository.php

@@ -12,11 +12,13 @@ public function find(int $id): StoredEvent;
 
     public function retrieveAll(?string $uuid = null): LazyCollection;
 
-    public function retrieveAllStartingFrom(int $startingFrom, ?string $uuid = null): LazyCollection;
+    public function retrieveAllStartingFrom(int $startingFrom, ?string $uuid = null, array $events = []): LazyCollection;
+
+    public function runForAllStartingFrom(int $startingFrom, callable|\Closure $function, int $chunkSize = 1000, ?string $uuid = null, array $events = []): bool;
 
     public function retrieveAllAfterVersion(int $aggregateVersion, string $aggregateUuid): LazyCollection;
 
-    public function countAllStartingFrom(int $startingFrom, ?string $uuid = null): int;
+    public function countAllStartingFrom(int $startingFrom, ?string $uuid = null, array $events = []): int;
 
     public function persist(ShouldBeStored $event, ?string $uuid = null): StoredEvent;