Fix Prooph projection manager reuse after reconnect

packages/PdoEventSourcing/src/Prooph/LazyProophProjectionManager.php

@@ -15,6 +15,7 @@
 use Ecotone\Messaging\MessageHeaders;
 use Ecotone\Messaging\Support\InvalidArgumentException;
 use Prooph\Common\Messaging\Message;
+use Prooph\EventStore\EventStore;
 use Prooph\EventStore\Exception\ProjectionNotFound;
 use Prooph\EventStore\Exception\RuntimeException;
 use Prooph\EventStore\Pdo\Projection\MariaDbProjectionManager;
@@ -36,8 +37,10 @@
  */
 class LazyProophProjectionManager implements ProjectionManager
 {
-    /** @var LazyProophProjectionManager[] */
+    /** @var ProjectionManager[] */
     private array $lazyInitializedProjectionManager = [];
+    /** @var EventStore[] */
+    private array $initializedEventStores = [];
 
     /**
      * @param array<string, ProjectionSetupConfiguration> $projectionSetupConfigurations
@@ -53,19 +56,24 @@ public function __construct(
 
     private function getProjectionManager(): ProjectionManager
     {
-        $context = $this->lazyProophEventStore->getContextName();
-        if (isset($this->lazyInitializedProjectionManager[$context])) {
-            return $this->lazyInitializedProjectionManager[$context];
+        if ($this->eventSourcingConfiguration->isInMemory()) {
+            return $this->eventSourcingConfiguration->getInMemoryProjectionManager();
         }
 
+        $context = $this->lazyProophEventStore->getContextName();
         $eventStore = $this->getLazyProophEventStore();
+        $innerEventStore = $eventStore->getEventStore();
+
+        if (isset($this->lazyInitializedProjectionManager[$context]) && $this->initializedEventStores[$context] === $innerEventStore) {
+            return $this->lazyInitializedProjectionManager[$context];
+        }
 
         $this->lazyInitializedProjectionManager[$context] = match ($eventStore->getEventStoreType()) {
-            LazyProophEventStore::EVENT_STORE_TYPE_POSTGRES => new PostgresProjectionManager($eventStore->getEventStore(), $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()),
-            LazyProophEventStore::EVENT_STORE_TYPE_MYSQL => new MySqlProjectionManager($eventStore->getEventStore(), $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()),
-            LazyProophEventStore::EVENT_STORE_TYPE_MARIADB => new MariaDbProjectionManager($eventStore->getEventStore(), $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()),
-            LazyProophEventStore::EVENT_STORE_TYPE_IN_MEMORY => $this->eventSourcingConfiguration->getInMemoryProjectionManager()
+            LazyProophEventStore::EVENT_STORE_TYPE_POSTGRES => new PostgresProjectionManager($innerEventStore, $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()),
+            LazyProophEventStore::EVENT_STORE_TYPE_MYSQL => new MySqlProjectionManager($innerEventStore, $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()),
+            LazyProophEventStore::EVENT_STORE_TYPE_MARIADB => new MariaDbProjectionManager($innerEventStore, $eventStore->getWrappedConnection(), $this->eventSourcingConfiguration->getEventStreamTableName(), $this->eventSourcingConfiguration->getProjectionsTable()),
         };
+        $this->initializedEventStores[$context] = $innerEventStore;
 
         return $this->lazyInitializedProjectionManager[$context];
     }

packages/PdoEventSourcing/tests/Integration/SynchronousEventDrivenProjectionTest.php

@@ -64,6 +64,48 @@ public function test_building_synchronous_event_driven_projection(): void
         self::assertEquals([['ticket_id' => '124', 'ticket_type' => 'info']], $ecotone->sendQueryWithRouting('getInProgressTickets'));
     }
 
+    public function test_synchronous_projection_uses_the_current_transaction_after_reconnecting(): void
+    {
+        $connection = $this->getConnection();
+        $ecotone = EcotoneLite::bootstrapFlowTestingWithEventStore(
+            containerOrAvailableServices: [new InProgressTicketList($connection), new TicketEventConverter(), DbalConnectionFactory::class => $this->getConnectionFactory()],
+            configuration: ServiceConfiguration::createWithDefaults()
+                ->withEnvironment('prod')
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::EVENT_SOURCING_PACKAGE]))
+                ->withNamespaces([
+                    'Test\\Ecotone\\EventSourcing\\Fixture\\Ticket',
+                    'Test\\Ecotone\\EventSourcing\\Fixture\\TicketWithSynchronousEventDrivenProjection',
+                ])
+                ->withExtensionObjects([
+                    EventSourcingConfiguration::createWithDefaults(),
+                ]),
+            pathToRootCatalog: __DIR__ . '/../../',
+            runForProductionEventStore: true
+        );
+
+        $ecotone->initializeProjection(InProgressTicketList::IN_PROGRESS_TICKET_PROJECTION);
+        $ecotone->sendCommand(new RegisterTicket('123', 'Johnny', 'alert'));
+        self::assertEquals([['ticket_id' => '123', 'ticket_type' => 'alert']], $ecotone->sendQueryWithRouting('getInProgressTickets'));
+
+        $connection->close();
+        $connection->beginTransaction();
+
+        try {
+            $ecotone->sendCommand(new CloseTicket('123'));
+
+            self::assertTrue($connection->isTransactionActive());
+            self::assertEquals([], $ecotone->sendQueryWithRouting('getInProgressTickets'));
+        } finally {
+            $connection->rollBack();
+        }
+
+        self::assertEquals([['ticket_id' => '123', 'ticket_type' => 'alert']], $ecotone->sendQueryWithRouting('getInProgressTickets'));
+
+        $ecotone->sendCommand(new CloseTicket('123'));
+
+        self::assertEquals([], $ecotone->sendQueryWithRouting('getInProgressTickets'));
+    }
+
     public function test_synchronous_event_driven_projection_should_be_called_before_standard_event_handlers(): void
     {
         $ecotone = EcotoneLite::bootstrapFlowTestingWithEventStore(