Feat: Async Batched Backfill (#606)

* feat: async batched projection backfills

* stream filtering

* backfill batch

* naming

* fixes

* async backfill

* fixes

* use stream filter

* fixes

* fixes

Author: dgafka

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/EcotoneConfiguration.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";

Monorepo/ExampleAppEventSourcing/EcotoneProjection/PriceChangeOverTimeProjectionWithEcotoneProjection.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
     ) {
     }
 }
+

…ng/src/Attribute/FromAggregateStream.php …urcing/Attribute/FromAggregateStream.phppackages/PdoEventSourcing/src/Attribute/FromAggregateStream.php renamed to packages/Ecotone/src/EventSourcing/Attribute/FromAggregateStream.php packages/PdoEventSourcing/src/Attribute/FromAggregateStream.php renamed to packages/Ecotone/src/EventSourcing/Attribute/FromAggregateStream.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(
     ) {
     }
 }
+

…entSourcing/src/Attribute/FromStream.php …c/EventSourcing/Attribute/FromStream.phppackages/PdoEventSourcing/src/Attribute/FromStream.php renamed to packages/Ecotone/src/EventSourcing/Attribute/FromStream.php packages/PdoEventSourcing/src/Attribute/FromStream.php renamed to packages/Ecotone/src/EventSourcing/Attribute/FromStream.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/Lite/Test/FlowTestSupport.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/Messaging/Config/ModuleClassList.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/ProjectionBackfill.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
+    )
+    {
+    }
+}