Feat: partition projections optimalization (#596)

* feat: partition projection optimalization

* refactor

* flaky tests

* fix phpstan

Author: dgafka

packages/Amqp/tests/AmqpMessagingTestCase.php

@@ -4,6 +4,7 @@
 
 use AMQPQueueException;
 use Ecotone\Amqp\Distribution\AmqpDistributionModule;
+use Ecotone\Enqueue\CachedConnectionFactory;
 use Enqueue\AmqpExt\AmqpConnectionFactory as AmqpExtConnection;
 use Enqueue\AmqpLib\AmqpConnectionFactory as AmqpLibConnection;
 use Interop\Amqp\AmqpConnectionFactory;
@@ -75,6 +76,9 @@ public static function getRabbitConnectionFactory(array $config = []): AmqpConne
 
     public function setUp(): void
     {
+        // Clear cached connection factories to prevent channel mode conflicts between tests
+        // (e.g., confirm mode vs transaction mode on the same channel)
+        CachedConnectionFactory::clearInstances();
         // Ensure cache directory is writable for tests
         $this->queueCleanUp();
     }

packages/Amqp/tests/Integration/SuccessTransactionTest.php

@@ -20,10 +20,6 @@ final class SuccessTransactionTest extends AmqpMessagingTestCase
 {
     public function test_order_is_placed_when_transaction_is_successful(): void
     {
-        if (getenv('AMQP_IMPLEMENTATION') === 'lib') {
-            $this->markTestSkipped('Transaction tests require Ext');
-        }
-
         $ecotone = $this->bootstrapFlowTesting(
             containerOrAvailableServices: [new OrderService(), ...$this->getConnectionFactoryReferences()],
             configuration: ServiceConfiguration::createWithDefaults()

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

@@ -67,8 +67,6 @@ public function prepare(Configuration $messagingConfiguration, array $extensionO
         /** @var array<string, array<string, string>> $components [projection name][component name][reference] */
         $components = [];
         foreach ($componentBuilders as $componentBuilder) {
-            $reference = Uuid::uuid4()->toString();
-            $moduleReferenceSearchService->store($reference, $componentBuilder);
             foreach ($projectionBuilders as $projectionBuilder) {
                 $projectionName = $projectionBuilder->projectionName();
                 foreach ([StreamSource::class, PartitionProvider::class, ProjectionStateStorage::class] as $component) {
@@ -79,6 +77,9 @@ public function prepare(Configuration $messagingConfiguration, array $extensionO
                                 . ' You can only register one component of each type per projection. Please check your configuration.'
                             );
                         }
+
+                        $reference = Uuid::uuid4()->toString();
+                        $moduleReferenceSearchService->store($reference, $componentBuilder);
                         $components[$projectionName][$component] = new Reference($reference);
                     }
                 }

packages/Ecotone/src/Projecting/EventStoreAdapter/EventStoreAdapterModule.php

@@ -72,7 +72,7 @@ public function getModuleExtensions(ServiceConfiguration $serviceConfiguration,
                 continue;
             }
 
-            $extensions[] = new EventStoreChannelAdapterProjectionBuilder($extensionObject);
+            $extensions[] = new EventStoreStreamingChannelAdapterBuilder($extensionObject);
         }
 
         return $extensions;

…StoreChannelAdapterProjectionBuilder.php …tStoreStreamingChannelAdapterBuilder.phppackages/Ecotone/src/Projecting/EventStoreAdapter/EventStoreChannelAdapterProjectionBuilder.php renamed to packages/Ecotone/src/Projecting/EventStoreAdapter/EventStoreStreamingChannelAdapterBuilder.php packages/Ecotone/src/Projecting/EventStoreAdapter/EventStoreChannelAdapterProjectionBuilder.php renamed to packages/Ecotone/src/Projecting/EventStoreAdapter/EventStoreStreamingChannelAdapterBuilder.php

@@ -18,7 +18,7 @@
  *
  * @internal
  */
-class EventStoreChannelAdapterProjectionBuilder implements ProjectionExecutorBuilder
+class EventStoreStreamingChannelAdapterBuilder implements ProjectionExecutorBuilder
 {
     public function __construct(
         private EventStoreChannelAdapter $channelAdapter

packages/Ecotone/src/Projecting/InMemory/InMemoryEventStoreStreamSource.php

@@ -7,6 +7,7 @@
 
 namespace Ecotone\Projecting\InMemory;
 
+use Ecotone\EventSourcing\EventStore\FieldType;
 use Ecotone\EventSourcing\EventStore\InMemoryEventStore;
 use Ecotone\EventSourcing\EventStore\MetadataMatcher;
 use Ecotone\EventSourcing\EventStore\Operator;
@@ -16,10 +17,14 @@
 
 class InMemoryEventStoreStreamSource implements StreamSource
 {
+    /**
+     * @param array<string> $eventNames Event names to filter by, empty array means no filtering
+     */
     public function __construct(
         private InMemoryEventStore $eventStore,
         private ?string $streamName = null,
-        private ?string $partitionHeader = null
+        private ?string $partitionHeader = null,
+        private array $eventNames = [],
     ) {
     }
 
@@ -38,20 +43,26 @@ public function load(?string $lastPosition, int $count, ?string $partitionKey =
                 continue;
             }
 
-            $metadataMatcher = null;
+            $metadataMatcher = new MetadataMatcher();
             if ($partitionKey !== null && $this->partitionHeader !== null) {
-                $metadataMatcher = (new MetadataMatcher())
+                $metadataMatcher = $metadataMatcher
                     ->withMetadataMatch($this->partitionHeader, Operator::EQUALS, $partitionKey);
             }
 
+            // Filter by event names if specified (optimization for partitioned projections)
+            if ($this->eventNames !== []) {
+                $metadataMatcher = $metadataMatcher
+                    ->withMetadataMatch('event_name', Operator::IN, $this->eventNames, FieldType::MESSAGE_PROPERTY);
+            }
+
             // Load all events from this stream (starting from position 1)
             $events = $this->eventStore->load($stream, 1, null, $metadataMatcher);
-            $allEvents = array_merge($allEvents, is_array($events) ? $events : iterator_to_array($events));
+            $allEvents = array_merge($allEvents, \is_array($events) ? $events : iterator_to_array($events));
         }
 
         // Slice based on global position
         $events = array_slice($allEvents, $from, $count);
-        $to = $from + count($events);
+        $to = $from + \count($events);
 
         return new StreamPage($events, (string) $to);
     }

packages/Ecotone/src/Projecting/InMemory/InMemoryEventStoreStreamSourceBuilder.php

@@ -16,17 +16,21 @@
 
 class InMemoryEventStoreStreamSourceBuilder implements ProjectionComponentBuilder
 {
+    /**
+     * @param array<string> $eventNames Event names to filter by, empty array means no filtering
+     */
     public function __construct(
         private ?array $projectionNames = null,
         private ?string $streamName = null,
-        private ?string $partitionHeader = null
+        private ?string $partitionHeader = null,
+        private array $eventNames = [],
     ) {
     }
 
     public function canHandle(string $projectionName, string $component): bool
     {
         return $component === StreamSource::class
-            && ($this->projectionNames === null || in_array($projectionName, $this->projectionNames, true));
+            && ($this->projectionNames === null || \in_array($projectionName, $this->projectionNames, true));
     }
 
     public function compile(MessagingContainerBuilder $builder): Definition|Reference
@@ -37,6 +41,7 @@ public function compile(MessagingContainerBuilder $builder): Definition|Referenc
                 Reference::to(InMemoryEventStore::class),
                 $this->streamName,
                 $this->partitionHeader,
+                $this->eventNames,
             ]
         );
     }

packages/Enqueue/src/CachedConnectionFactory.php

@@ -51,6 +51,16 @@ public function reconnect(): void
         $this->cachedContext = [];
     }
 
+    /**
+     * Clear all cached connection factory instances.
+     * This is useful in tests to ensure clean state between test runs,
+     * especially when switching between different channel modes (confirm vs tx).
+     */
+    public static function clearInstances(): void
+    {
+        self::$instances = [];
+    }
+
     public function getConsumer(Destination $destination): Consumer
     {
         return $this->createContext()->createConsumer($destination);