fix(symfony): reset the Mercure publisher between requests

PublishMercureUpdatesListener collects the created, updated and deleted
resources in onFlush and publishes them in postFlush. When a flush fails,
postFlush is never dispatched and the collected objects stay in the
listener, so a long-running worker could publish them during another
request. Make the listener resettable and tag it kernel.reset.

Author: KevinMartinsDev

src/Symfony/Bundle/Resources/config/doctrine_odm_mercure_publisher.php

@@ -33,5 +33,6 @@
             '%api_platform.mercure.include_type%',
         ])
         ->tag('doctrine_mongodb.odm.event_listener', ['event' => 'onFlush'])
-        ->tag('doctrine_mongodb.odm.event_listener', ['event' => 'postFlush']);
+        ->tag('doctrine_mongodb.odm.event_listener', ['event' => 'postFlush'])
+        ->tag('kernel.reset', ['method' => 'reset', 'on_invalid' => 'ignore']);
 };

src/Symfony/Bundle/Resources/config/doctrine_orm_mercure_publisher.php

@@ -33,5 +33,6 @@
             '%api_platform.mercure.include_type%',
         ])
         ->tag('doctrine.event_listener', ['event' => 'onFlush'])
-        ->tag('doctrine.event_listener', ['event' => 'postFlush']);
+        ->tag('doctrine.event_listener', ['event' => 'postFlush'])
+        ->tag('kernel.reset', ['method' => 'reset', 'on_invalid' => 'ignore']);
 };

src/Symfony/Doctrine/EventListener/PublishMercureUpdatesListener.php

@@ -36,13 +36,14 @@
 use Symfony\Component\Mercure\Update;
 use Symfony\Component\Messenger\MessageBusInterface;
 use Symfony\Component\Serializer\SerializerInterface;
+use Symfony\Contracts\Service\ResetInterface;
 
 /**
  * Publishes resources updates to the Mercure hub.
  *
  * @author Kévin Dunglas <dunglas@gmail.com>
  */
-final class PublishMercureUpdatesListener
+final class PublishMercureUpdatesListener implements ResetInterface
 {
     use DispatchTrait;
     use ResourceClassInfoTrait;
@@ -149,7 +150,7 @@ public function postFlush(): void
         }
     }
 
-    private function reset(): void
+    public function reset(): void
     {
         $this->createdObjects = [];
         $this->updatedObjects = [];

src/Symfony/Tests/Bundle/DependencyInjection/ApiPlatformExtensionTest.php

@@ -470,6 +470,16 @@ public function testMcpProviderChainIsNotSecuredWithoutSecurityBundle(): void
         $this->assertContainerHasService('api_platform.mcp.state_provider.parameter_validator');
     }
 
+    public function testMercurePublisherIsReset(): void
+    {
+        $config = self::DEFAULT_CONFIG;
+        $config['api_platform']['doctrine']['enabled'] = true;
+        $config['api_platform']['mercure']['enabled'] = true;
+        (new ApiPlatformExtension())->load($config, $this->container);
+
+        $this->assertSame([['method' => 'reset', 'on_invalid' => 'ignore']], $this->container->getDefinition('api_platform.doctrine.orm.listener.mercure.publish')->getTag('kernel.reset'));
+    }
+
     public function testItRegistersMetadataConfiguration(): void
     {
         $config = self::DEFAULT_CONFIG;

src/Symfony/Tests/Doctrine/EventListener/PublishMercureUpdatesListenerTest.php

@@ -172,6 +172,59 @@ public function testPublishUpdate(): void
         $this->assertEquals([null, null, null, null, null, 10, null], $retry);
     }
 
+    public function testResetDropsTheUpdatesCollectedByAFailedFlush(): void
+    {
+        $failed = new Dummy();
+        $failed->setId(1);
+        $published = new Dummy();
+        $published->setId(2);
+
+        $resourceClassResolverProphecy = $this->prophesize(ResourceClassResolverInterface::class);
+        $resourceClassResolverProphecy->getResourceClass(Argument::type(Dummy::class))->willReturn(Dummy::class);
+        $resourceClassResolverProphecy->isResourceClass(Dummy::class)->willReturn(true);
+
+        $iriConverterProphecy = $this->prophesize(IriConverterInterface::class);
+        $iriConverterProphecy->getIriFromResource($failed, Argument::cetera())->shouldNotBeCalled();
+        $iriConverterProphecy->getIriFromResource($published, UrlGeneratorInterface::ABS_URL, Argument::any())->willReturn('http://example.com/dummies/2')->shouldBeCalled();
+
+        $resourceMetadataFactoryProphecy = $this->prophesize(ResourceMetadataCollectionFactoryInterface::class);
+        $resourceMetadataFactoryProphecy->create(Dummy::class)->willReturn(new ResourceMetadataCollection(Dummy::class, [(new ApiResource())->withOperations(new Operations([
+            'get' => (new Get())->withShortName('Dummy')->withMercure(['hub' => 'managed', 'enable_async_update' => false]),
+        ]))]));
+
+        $serializerProphecy = $this->prophesize(SerializerInterface::class);
+        $serializerProphecy->serialize($published, 'jsonld', [])->willReturn('2');
+
+        $topics = [];
+        $managedHub = $this->createMockHub(static function (Update $update) use (&$topics): string {
+            $topics = array_merge($topics, $update->getTopics());
+
+            return 'id';
+        });
+
+        $listener = new PublishMercureUpdatesListener(
+            $resourceClassResolverProphecy->reveal(),
+            $iriConverterProphecy->reveal(),
+            $resourceMetadataFactoryProphecy->reveal(),
+            $serializerProphecy->reveal(),
+            ['jsonld' => ['application/ld+json']],
+            null,
+            new HubRegistry($this->createStub(HubInterface::class), [
+                'managed' => $managedHub,
+            ]),
+        );
+
+        // the commit of the first flush fails: Doctrine dispatches onFlush, but never postFlush
+        $listener->onFlush($this->createOnFlushEventArgs([$failed]));
+        // between two requests (or two messages), the services resetter resets the listener
+        $listener->reset();
+
+        $listener->onFlush($this->createOnFlushEventArgs([$published]));
+        $listener->postFlush();
+
+        $this->assertSame(['http://example.com/dummies/2'], $topics);
+    }
+
     public function testPublishUpdateMultipleTopicsUsingExpressionLanguage(): void
     {
         $mercure = [
@@ -444,6 +497,19 @@ public function testPublishUpdateWithMultipleResources(): void
         $this->assertEquals('{"public":1}', $data[1]);
     }
 
+    private function createOnFlushEventArgs(array $insertions): OnFlushEventArgs
+    {
+        $uowProphecy = $this->prophesize(UnitOfWork::class);
+        $uowProphecy->getScheduledEntityInsertions()->willReturn($insertions);
+        $uowProphecy->getScheduledEntityUpdates()->willReturn([]);
+        $uowProphecy->getScheduledEntityDeletions()->willReturn([]);
+
+        $emProphecy = $this->prophesize(EntityManagerInterface::class);
+        $emProphecy->getUnitOfWork()->willReturn($uowProphecy->reveal());
+
+        return new OnFlushEventArgs($emProphecy->reveal());
+    }
+
     private function createMockHub(callable $callable): HubInterface
     {
         return new MockHub('https://mercure.demo/.well-known/mercure', new StaticTokenProvider('x'), $callable);