Projecting improvements 1 (#522)

* fix DbalProjectionStateStorage.php for manager registry connection factory

* init all projections in one command

* better gap timeout processing

Author: jlabedo

packages/Ecotone/src/Messaging/Scheduling/Duration.php

@@ -29,6 +29,11 @@ public static function seconds(int|float $seconds): self
         return new self((int) round($seconds * 1_000_000));
     }
 
+    public static function minutes(int|float $seconds): self
+    {
+        return new self((int) round($seconds * 60 * 1_000_000));
+    }
+
     public static function zero(): self
     {
         return new self(0);

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

@@ -8,6 +8,7 @@
 namespace Ecotone\Projecting\Config;
 
 use Ecotone\Messaging\Attribute\ConsoleCommand;
+use Ecotone\Messaging\Attribute\ConsoleParameterOption;
 use Ecotone\Projecting\ProjectionRegistry;
 use InvalidArgumentException;
 
@@ -18,12 +19,21 @@ public function __construct(private ProjectionRegistry $registry)
     }
 
     #[ConsoleCommand('ecotone:projection:init')]
-    public function initProjection(string $name): void
+    public function initProjection(?string $name = null, #[ConsoleParameterOption] bool $all = false): void
     {
-        if (! $this->registry->has($name)) {
-            throw new InvalidArgumentException("There is no projection with name {$name}");
+        if ($name === null) {
+            if (! $all) {
+                throw new InvalidArgumentException('You need to provide projection name or use --all option');
+            }
+            foreach ($this->registry->projectionNames() as $projection) {
+                $this->registry->get($projection)->init();
+            }
+        } else {
+            if (! $this->registry->has($name)) {
+                throw new InvalidArgumentException("There is no projection with name {$name}");
+            }
+            $this->registry->get($name)->init();
         }
-        $this->registry->get($name)->init();
     }
 
     #[ConsoleCommand('ecotone:projection:backfill')]

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

@@ -36,4 +36,9 @@ public function get(string $id): ProjectingManager
 
         return $this->projectionManagers[$id];
     }
+
+    public function projectionNames(): iterable
+    {
+        return array_keys($this->projectionManagers);
+    }
 }

packages/Ecotone/src/Projecting/ProjectingManager.php

@@ -19,6 +19,7 @@ public function __construct(
         private PartitionProvider      $partitionProvider,
         private string                 $projectionName,
         private int                    $batchSize = 1000,
+        private bool                   $autoInit = true,
     ) {
         if ($batchSize < 1) {
             throw new InvalidArgumentException('Batch size must be at least 1');
@@ -28,7 +29,9 @@ public function __construct(
     // This is the method that is linked to the event bus routing channel
     public function execute(?string $partitionKey = null): void
     {
-        $this->init();
+        if ($this->autoInit) {
+            $this->init();
+        }
 
         do {
             $transaction = $this->projectionStateStorage->beginTransaction();

packages/Ecotone/src/Projecting/ProjectionRegistry.php

@@ -12,4 +12,6 @@
 interface ProjectionRegistry extends ContainerInterface
 {
     public function get(string $id): ProjectingManager;
+    /** @return iterable<string> */
+    public function projectionNames(): iterable;
 }

packages/PdoEventSourcing/src/Projecting/PartitionState/DbalProjectionStateStorage.php

@@ -15,6 +15,7 @@
 use Ecotone\Projecting\Transaction;
 use Enqueue\Dbal\DbalConnectionFactory;
 
+use Enqueue\Dbal\ManagerRegistryConnectionFactory;
 use function json_decode;
 use function json_encode;
 
@@ -28,7 +29,7 @@ class DbalProjectionStateStorage implements ProjectionStateStorage
     private bool $initialized = false;
 
     public function __construct(
-        DbalConnectionFactory $connectionFactory,
+        DbalConnectionFactory|ManagerRegistryConnectionFactory $connectionFactory,
         private string        $stateTable = 'ecotone_projection_state',
     ) {
         $this->connection = $connectionFactory->createContext()->getDbalConnection();

packages/PdoEventSourcing/src/Projecting/StreamSource/EventStoreGlobalStreamSource.php

@@ -25,7 +25,7 @@ public function __construct(
         private EcotoneClockInterface  $clock,
         private string          $streamName,
         private int             $maxGapOffset = 5_000,
-        private ?Duration            $gapTimeout = null,
+        private ?Duration       $gapTimeout = null,
     ) {
     }
 
@@ -39,6 +39,7 @@ public function load(?string $lastPosition, int $count, ?string $partitionKey =
         } else {
             $eventsInGaps = $this->eventStore->load(
                 $this->streamName,
+                count: count($tracking->getGaps()),
                 metadataMatcher: (new MetadataMatcher())
                     ->withMetadataMatch('no', Operator::IN(), $tracking->getGaps(), FieldType::MESSAGE_PROPERTY()),
                 deserialize: false,
@@ -55,9 +56,13 @@ public function load(?string $lastPosition, int $count, ?string $partitionKey =
 
         $allEvents = [...$eventsInGaps, ...$events];
 
+        $now = $this->clock->now();
+        $cutoffTimestamp = $this->gapTimeout ? $now->sub($this->gapTimeout)->getTimestamp() : 0;
         foreach ($allEvents as $event) {
             $position = $event->getMetadata()['_position'] ?? throw new RuntimeException('Event does not have a position');
-            $tracking->advanceTo((int) $position);
+            $timestamp = $event->getMetadata()['timestamp'] ?? throw new RuntimeException('Event does not have a timestamp');
+            $insertGaps = $timestamp > $cutoffTimestamp;
+            $tracking->advanceTo((int) $position, $insertGaps);
         }
 
         $tracking->cleanByMaxOffset($this->maxGapOffset);
@@ -77,31 +82,41 @@ private function cleanGapsByTimeout(GapAwarePosition $tracking): void
             return;
         }
 
-        $minGap = min($gaps);
-        $maxGap = max($gaps);
+        $minGap = $gaps[0];
+        $maxGap = $gaps[count($gaps) - 1];
 
         // Query interleaved events in the gap range
         $interleavedEvents = $this->eventStore->load(
             $this->streamName,
+            count: count($gaps),
             metadataMatcher: (new MetadataMatcher())
                 ->withMetadataMatch('no', Operator::GREATER_THAN_EQUALS(), $minGap, FieldType::MESSAGE_PROPERTY())
-                ->withMetadataMatch('no', Operator::LOWER_THAN_EQUALS(), $maxGap, FieldType::MESSAGE_PROPERTY()),
+                ->withMetadataMatch('no', Operator::LOWER_THAN_EQUALS(), $maxGap + 1, FieldType::MESSAGE_PROPERTY()),
             deserialize: false,
         );
 
-        $timestampThreshold = $this->clock->now()->unixTime()->sub($this->gapTimeout)->inSeconds();
+        $timestampThreshold = $this->clock->now()->sub($this->gapTimeout)->unixTime()->inSeconds();
 
         // Find the highest position with timestamp < timeThreshold
         $cutoffPosition = $minGap; // default: keep all gaps
         foreach ($interleavedEvents as $event) {
             $metadata = $event->getMetadata();
-            $position = $metadata['_position'] ?? null;
+            $interleavedEventPosition = ((int)$metadata['_position']) ?? null;
             $timestamp = $metadata['timestamp'] ?? null;
 
-            if ($position !== null && $timestamp !== null) {
-                if ($timestamp < $timestampThreshold && $position > $cutoffPosition) {
-                    $cutoffPosition = $position + 1; // Remove gaps below this position
-                }
+            if ($interleavedEventPosition === null || $timestamp === null) {
+                break;
+            }
+            if ($timestamp > $timestampThreshold) {
+                // Event is recent, do not remove any gaps below this position
+                break;
+            }
+            if (in_array($interleavedEventPosition, $gaps, true)) {
+                // This position is a gap, stop cleaning
+                break;
+            }
+            if ($timestamp < $timestampThreshold && $interleavedEventPosition > $cutoffPosition) {
+                $cutoffPosition = $interleavedEventPosition + 1; // Remove gaps below this position
             }
         }
 

packages/PdoEventSourcing/src/Projecting/StreamSource/EventStoreGlobalStreamSourceBuilder.php

@@ -11,6 +11,7 @@
 use Ecotone\Messaging\Config\Container\Definition;
 use Ecotone\Messaging\Config\Container\MessagingContainerBuilder;
 use Ecotone\Messaging\Config\Container\Reference;
+use Ecotone\Messaging\Scheduling\Duration;
 use Ecotone\Messaging\Scheduling\EcotoneClockInterface;
 use Ecotone\Projecting\Config\ProjectionComponentBuilder;
 use Ecotone\Projecting\StreamSource;
@@ -36,6 +37,8 @@ public function compile(MessagingContainerBuilder $builder): Definition|Referenc
                 Reference::to(EventStore::class),
                 Reference::to(EcotoneClockInterface::class),
                 $this->streamName,
+                5_000,
+                new Definition(Duration::class, [60], 'seconds')
             ],
         );
     }

packages/PdoEventSourcing/src/Projecting/StreamSource/GapAwarePosition.php

@@ -74,8 +74,18 @@ public function getGaps(): array
         return $this->gaps;
     }
 
-    public function advanceTo(int $position): void
+    public function advanceTo(int $position, bool $processGaps = true): void
     {
+        if ($processGaps === false) {
+            if ($position > $this->position) {
+                $this->position = $position;
+            } elseif ($position <= $this->position) {
+                throw new InvalidArgumentException('Cannot advance to a position less than or equal to the current position. Current position: ' . $this->position . ', new position: ' . $position);
+            }
+            return;
+        }
+
+        // With gap processing
         if ($position === $this->position + 1) {
             $this->position++;
         } elseif (in_array($position, $this->gaps, true)) {

packages/PdoEventSourcing/tests/Projecting/GapAwarePositionIntegrationTest.php

@@ -131,7 +131,6 @@ public function test_gap_timeout_cleaning(): void
             self::$eventStore,
             self::$clock,
             Ticket::STREAM_NAME,
-            maxGapOffset: 1000,
             gapTimeout: Duration::seconds(5)
         );
 
@@ -142,15 +141,21 @@ public function test_gap_timeout_cleaning(): void
         $tracking = GapAwarePosition::fromString($result->lastPosition);
         self::assertSame([2, 4, 6], $tracking->getGaps());
 
-        // Delay 2 more seconds to exceed timeout for first gaps
+        // Delay 2 more seconds to exceed timeout for first gaps (6 seconds since insertion)
         self::$clock->sleep(Duration::seconds(2));
 
         // Execute
         $result = $streamSource->load(null, 100);
 
-        // Verify: Gaps 2, 7 should be removed (old timestamps), gaps 5, 9 should remain (recent timestamps)
+        // Verify: Gaps 2, 4 should be removed (old timestamps), gap 6 should remain (recent timestamps)
         $newTracking = GapAwarePosition::fromString($result->lastPosition);
         self::assertSame([6], $newTracking->getGaps());
+
+        // Delay 3 more second to exceed timeout for all gaps (6 seconds since insertion of the last event)
+        self::$clock->sleep(Duration::seconds(4));
+        $result = $streamSource->load($result->lastPosition, 100);
+        $newTracking = GapAwarePosition::fromString($result->lastPosition);
+        self::assertSame([], $newTracking->getGaps());
     }
 
     public function test_gap_cleaning_noop_when_no_gaps(): void