Feat: Async Batched Backfill

Monorepo/Benchmark/ProjectingBenchmark.php

@@ -174,7 +174,7 @@ public function bench_ecotone_projection_backfill(): void
         Assert::assertEquals([],
             self::$ecotone->getQueryBus()->sendWithRouting('product.getPriceChange', self::$expectedProductIds[0])
         );
-        $projectionManager->backfill();
+        $projectionManager->prepareBackfill();
         Assert::assertEquals([
             new PriceChange(100, 0),
             new PriceChange(120, 20),

Monorepo/ExampleAppEventSourcing/EcotoneProjection/PriceChangeOverTimeProjectionWithEcotoneProjection.php

@@ -4,14 +4,18 @@
  */
 namespace Monorepo\ExampleAppEventSourcing\EcotoneProjection;
 
+use Ecotone\EventSourcing\Attribute\FromAggregateStream;
 use Ecotone\EventSourcing\Attribute\ProjectionDelete;
 use Ecotone\Modelling\Attribute\EventHandler;
 use Ecotone\Modelling\Attribute\QueryHandler;
+use Ecotone\Projecting\Attribute\ProjectionV2;
 use Monorepo\ExampleAppEventSourcing\Common\Event\PriceWasChanged;
 use Monorepo\ExampleAppEventSourcing\Common\Event\ProductWasRegistered;
 use Monorepo\ExampleAppEventSourcing\Common\PriceChange;
+use Monorepo\ExampleAppEventSourcing\Common\Product;
 
-#[\Ecotone\Projecting\Attribute\ProjectionV2("price_change_over_time")]
+#[ProjectionV2("price_change_over_time")]
+#[FromAggregateStream(Product::class)]
 class PriceChangeOverTimeProjectionWithEcotoneProjection
 {
     public const NAME = "price_change_over_time";

packages/PdoEventSourcing/src/Attribute/FromAggregateStream.php → packages/Ecotone/src/EventSourcing/Attribute/FromAggregateStream.php

@@ -1,40 +1,37 @@
 <?php
 
+/*
+ * licence Enterprise
+ */
 declare(strict_types=1);
 
 namespace Ecotone\EventSourcing\Attribute;
 
 use Attribute;
 use Ecotone\EventSourcing\EventStore;
-use Ecotone\Modelling\Attribute\EventSourcingAggregate;
 
-/*
+/**
  * Configures a projection to read from an aggregate's event stream.
- * Automatically reads Stream and AggregateType attributes from the aggregate class.
- *
- * This simplifies projection configuration by avoiding duplication of stream
- * and aggregate type configuration that is already defined on the aggregate.
+ * Automatically resolves Stream and AggregateType from the aggregate class.
  *
  * Example usage:
  * ```php
  * #[ProjectionV2('order_list')]
- * #[AggregateStream(Order::class)]
+ * #[FromAggregateStream(Order::class)]
  * class OrderListProjection { ... }
  * ```
- *
- * licence Enterprise
  */
 #[Attribute(Attribute::TARGET_CLASS | Attribute::IS_REPEATABLE)]
-class FromAggregateStream
+readonly class FromAggregateStream
 {
     /**
-     * @param class-string $aggregateClass The aggregate class to read Stream and AggregateType from.
+     * @param class-string $aggregateClass The aggregate class to read stream info from.
      *                                      Must be an EventSourcingAggregate.
-     * @param string $eventStoreReferenceName Reference name for the event store
      */
     public function __construct(
-        public readonly string $aggregateClass,
-        public readonly string $eventStoreReferenceName = EventStore::class
+        public string $aggregateClass,
+        public string $eventStoreReferenceName = EventStore::class
     ) {
     }
 }
+

packages/PdoEventSourcing/src/Attribute/FromStream.php → packages/Ecotone/src/EventSourcing/Attribute/FromStream.php

@@ -10,6 +10,9 @@
 use Attribute;
 use Ecotone\EventSourcing\EventStore;
 
+/**
+ * Configures a projection to read from a specific event stream.
+ */
 #[Attribute(Attribute::TARGET_CLASS | Attribute::IS_REPEATABLE)]
 readonly class FromStream
 {
@@ -20,3 +23,4 @@ public function __construct(
     ) {
     }
 }
+

packages/Ecotone/src/Lite/Test/FlowTestSupport.php

@@ -237,7 +237,7 @@ public function triggerProjection(string|array $projectionName): self
 
         foreach ($projectionName as $name) {
             if ($this->getGateway(ProjectionRegistry::class)->has($name)) {
-                $this->getGateway(ProjectionRegistry::class)->get($name)->backfill();
+                $this->getGateway(ProjectionRegistry::class)->get($name)->prepareBackfill();
             } else {
                 $this->getGateway(ProjectionManager::class)->triggerProjection($name);
             }

packages/Ecotone/src/Messaging/Config/ModuleClassList.php

@@ -68,6 +68,7 @@
 use Ecotone\Projecting\Config\ProjectingAttributeModule;
 use Ecotone\Projecting\Config\ProjectingConsoleCommands;
 use Ecotone\Projecting\Config\ProjectingModule;
+use Ecotone\Projecting\Config\StreamFilterRegistryModule;
 use Ecotone\Projecting\EventStoreAdapter\EventStoreAdapterModule;
 use Ecotone\Redis\Configuration\RedisMessageConsumerModule;
 use Ecotone\Redis\Configuration\RedisMessagePublisherModule;
@@ -115,6 +116,7 @@ class ModuleClassList
         EventSourcedRepositoryModule::class,
         ProjectingModule::class,
         ProjectingAttributeModule::class,
+        StreamFilterRegistryModule::class,
         EventStoreAdapterModule::class,
 
         /** Attribute based configurations */

packages/Ecotone/src/Projecting/Attribute/ProjectionBackfill.php

@@ -0,0 +1,38 @@
+<?php
+
+/*
+ * licence Enterprise
+ */
+declare(strict_types=1);
+
+namespace Ecotone\Projecting\Attribute;
+
+use Attribute;
+use InvalidArgumentException;
+
+/**
+ * Configure projection backfill settings.
+ * This attribute controls how partitions are batched during backfill operations.
+ */
+#[Attribute(Attribute::TARGET_CLASS)]
+class ProjectionBackfill
+{
+    public function __construct(
+        /**
+         * Number of partitions to process in a single batch during backfill.
+         * Must be at least 1.
+         */
+        public readonly int $backfillPartitionBatchSize = 100,
+        /**
+         * Async channel name for backfill operations.
+         * When set, backfill batches are sent to this channel first, then routed to the backfill handler.
+         * When null, backfill executes synchronously.
+         */
+        public readonly ?string $asyncChannelName = null,
+    ) {
+        if ($this->backfillPartitionBatchSize < 1) {
+            throw new InvalidArgumentException('Backfill partition batch size must be at least 1');
+        }
+    }
+}
+

packages/Ecotone/src/Projecting/Attribute/ProjectionExecution.php

@@ -0,0 +1,25 @@
+<?php
+
+/*
+ * licence Enterprise
+ */
+declare(strict_types=1);
+
+namespace Ecotone\Projecting\Attribute;
+
+use Attribute;
+
+#[Attribute(Attribute::TARGET_CLASS)]
+class ProjectionExecution
+{
+    public function __construct(
+        /**
+         * Configure the batch size for loading events during projection execution.
+         * * This controls how many events are loaded from the stream in a single batch.
+
+         */
+        public readonly int $eventLoadingBatchSize
+    )
+    {
+    }
+}

packages/Ecotone/src/Projecting/BackfillExecutorHandler.php

@@ -0,0 +1,56 @@
+<?php
+
+/*
+ * licence Enterprise
+ */
+declare(strict_types=1);
+
+namespace Ecotone\Projecting;
+
+use Ecotone\Messaging\Endpoint\Interceptor\TerminationListener;
+
+/**
+ * Handles execution of projection backfill batches.
+ * This handler is invoked via MessagingEntrypoint to execute backfill operations
+ * for a given projection with specified limit and offset parameters.
+ */
+class BackfillExecutorHandler
+{
+    public const BACKFILL_EXECUTOR_CHANNEL = 'ecotone.projection.backfill.executor';
+
+    public function __construct(
+        private ProjectionRegistry $projectionRegistry,
+        private TerminationListener $terminationListener,
+    ) {
+    }
+
+    /**
+     * Execute backfill for a specific partition batch.
+     *
+     * @param string $projectionName The name of the projection to backfill
+     * @param int|null $limit The maximum number of partitions to process in this batch (null for unlimited)
+     * @param int $offset The offset to start from
+     * @param string $streamName The stream name to filter partitions
+     * @param string|null $aggregateType The aggregate type to filter partitions (optional)
+     * @param string $eventStoreReferenceName The event store reference name
+     */
+    public function executeBackfillBatch(
+        string $projectionName,
+        ?int $limit = null,
+        int $offset = 0,
+        string $streamName = '',
+        ?string $aggregateType = null,
+        string $eventStoreReferenceName = '',
+    ): void {
+        $projectingManager = $this->projectionRegistry->get($projectionName);
+        $streamFilter = new StreamFilter($streamName, $aggregateType, $eventStoreReferenceName);
+
+        foreach ($projectingManager->getPartitionProvider()->partitions($streamFilter, $limit, $offset) as $partition) {
+            $projectingManager->execute($partition, true);
+            if ($this->terminationListener->shouldTerminate()) {
+                break;
+            }
+        }
+    }
+}
+

packages/Ecotone/src/Projecting/Config/EcotoneProjectionExecutorBuilder.php

@@ -25,7 +25,8 @@
 
 class EcotoneProjectionExecutorBuilder implements ProjectionExecutorBuilder
 {
-    private const DEFAULT_BATCH_SIZE = 1_000;
+    private const DEFAULT_EVENT_LOADING_BATCH_SIZE = 1_000;
+    private const DEFAULT_BACKFILL_PARTITION_BATCH_SIZE = 100;
 
     /**
      * @param AnnotatedDefinition[] $projectionEventHandlers
@@ -41,7 +42,9 @@ public function __construct(
         private ?string $flushChannel = null,
         private array   $projectionEventHandlers = [],
         private ?string $asyncChannelName = null,
-        private ?int    $batchSize = null,
+        private ?int    $eventLoadingBatchSize = null,
+        private ?int    $backfillPartitionBatchSize = null,
+        private ?string $backfillAsyncChannelName = null,
     ) {
         if ($this->partitionHeader && ! $this->automaticInitialization) {
             throw new ConfigurationException("Cannot set partition header for projection {$this->projectionName} with automatic initialization disabled");
@@ -93,9 +96,19 @@ public function automaticInitialization(): bool
         return $this->automaticInitialization;
     }
 
-    public function batchSize(): int
+    public function eventLoadingBatchSize(): int
     {
-        return $this->batchSize ?? self::DEFAULT_BATCH_SIZE;
+        return $this->eventLoadingBatchSize ?? self::DEFAULT_EVENT_LOADING_BATCH_SIZE;
+    }
+
+    public function backfillPartitionBatchSize(): int
+    {
+        return $this->backfillPartitionBatchSize ?? self::DEFAULT_BACKFILL_PARTITION_BATCH_SIZE;
+    }
+
+    public function backfillAsyncChannelName(): ?string
+    {
+        return $this->backfillAsyncChannelName;
     }
 
     public function compile(MessagingContainerBuilder $builder): Definition|Reference

packages/Ecotone/src/Projecting/Config/ProjectingAttributeModule.php

@@ -34,7 +34,8 @@
 use Ecotone\Modelling\Attribute\NamedEvent;
 use Ecotone\Projecting\Attribute\Partitioned;
 use Ecotone\Projecting\Attribute\Polling;
-use Ecotone\Projecting\Attribute\ProjectionBatchSize;
+use Ecotone\Projecting\Attribute\ProjectionBackfill;
+use Ecotone\Projecting\Attribute\ProjectionExecution;
 use Ecotone\Projecting\Attribute\ProjectionDeployment;
 use Ecotone\Projecting\Attribute\ProjectionFlush;
 use Ecotone\Projecting\Attribute\ProjectionV2;
@@ -77,7 +78,8 @@ public static function create(AnnotationFinder $annotationRegistrationService, I
         $eventStreamingProjections = [];
         foreach ($annotationRegistrationService->findAnnotatedClasses(ProjectionV2::class) as $projectionClassName) {
             $projectionAttribute = $annotationRegistrationService->getAttributeForClass($projectionClassName, ProjectionV2::class);
-            $batchSizeAttribute = $annotationRegistrationService->findAttributeForClass($projectionClassName, ProjectionBatchSize::class);
+            $batchSizeAttribute = $annotationRegistrationService->findAttributeForClass($projectionClassName, ProjectionExecution::class);
+            $backfillAttribute = $annotationRegistrationService->findAttributeForClass($projectionClassName, ProjectionBackfill::class);
             $pollingAttribute = $annotationRegistrationService->findAttributeForClass($projectionClassName, Polling::class);
             $streamingAttribute = $annotationRegistrationService->findAttributeForClass($projectionClassName, Streaming::class);
             $projectionDeployment = $annotationRegistrationService->findAttributeForClass($projectionClassName, ProjectionDeployment::class);
@@ -88,7 +90,16 @@ public static function create(AnnotationFinder $annotationRegistrationService, I
             $automaticInitialization = self::resolveAutomaticInitialization($partitionAttribute, $projectionDeployment);
             $isLive = $projectionDeployment?->live ?? true;
 
-            $projectionBuilder = new EcotoneProjectionExecutorBuilder($projectionAttribute->name, $partitionHeaderName, $automaticInitialization, $isLive, $namedEvents, batchSize: $batchSizeAttribute?->batchSize);
+            $projectionBuilder = new EcotoneProjectionExecutorBuilder(
+                $projectionAttribute->name,
+                $partitionHeaderName,
+                $automaticInitialization,
+                $isLive,
+                $namedEvents,
+                eventLoadingBatchSize: $batchSizeAttribute?->eventLoadingBatchSize,
+                backfillPartitionBatchSize: $backfillAttribute?->backfillPartitionBatchSize,
+                backfillAsyncChannelName: $backfillAttribute?->asyncChannelName,
+            );
 
             $asynchronousChannelName = self::getProjectionAsynchronousChannel($annotationRegistrationService, $projectionClassName);
             $isPolling = $pollingAttribute !== null;

packages/Ecotone/src/Projecting/Config/ProjectingConsoleCommands.php

@@ -42,7 +42,7 @@ public function backfillProjection(string $name): void
         if (! $this->registry->has($name)) {
             throw new InvalidArgumentException("There is no projection with name {$name}");
         }
-        $this->registry->get($name)->backfill();
+        $this->registry->get($name)->prepareBackfill();
     }
 
     #[ConsoleCommand('ecotone:projection:delete')]