Created event listener that notifies mercure hub for new visits

composer.json

@@ -50,7 +50,7 @@
         "predis/predis": "^1.1",
         "pugx/shortid-php": "^0.5",
         "ramsey/uuid": "^3.9",
-        "shlinkio/shlink-common": "dev-master#3178fd18ce729d1dd63f4fb54f6a3696a11fef3f as 3.1.0",
+        "shlinkio/shlink-common": "dev-master#e659cf9d9b5b3b131419e2f55f2e595f562baafc as 3.1.0",
         "shlinkio/shlink-config": "^1.0",
         "shlinkio/shlink-event-dispatcher": "^1.4",
         "shlinkio/shlink-installer": "^4.3.2",

config/autoload/mercure.global.php

@@ -4,6 +4,8 @@ declare(strict_types=1);
 
 use Laminas\ServiceManager\Proxy\LazyServiceFactory;
 use Shlinkio\Shlink\Common\Mercure\LcobucciJwtProvider;
+use Symfony\Component\Mercure\Publisher;
+use Symfony\Component\Mercure\PublisherInterface;
 
 return [
 
@@ -20,10 +22,14 @@ return [
             LcobucciJwtProvider::class => [
                 LazyServiceFactory::class,
             ],
+            Publisher::class => [
+                LazyServiceFactory::class,
+            ],
         ],
         'lazy_services' => [
             'class_map' => [
                 LcobucciJwtProvider::class => LcobucciJwtProvider::class,
+                Publisher::class => PublisherInterface::class,
             ],
         ],
     ],

module/Core/config/dependencies.config.php

@@ -38,6 +38,8 @@ return [
             Action\QrCodeAction::class => ConfigAbstractFactory::class,
 
             Resolver\PersistenceDomainResolver::class => ConfigAbstractFactory::class,
+
+            Mercure\MercureUpdatesGenerator::class => ConfigAbstractFactory::class,
         ],
     ],
 
@@ -83,6 +85,8 @@ return [
         ],
 
         Resolver\PersistenceDomainResolver::class => ['em'],
+
+        Mercure\MercureUpdatesGenerator::class => ['config.url_shortener.domain'],
     ],
 
 ];

module/Core/config/event_dispatcher.config.php

@@ -8,12 +8,14 @@ use Laminas\ServiceManager\AbstractFactory\ConfigAbstractFactory;
 use Psr\EventDispatcher\EventDispatcherInterface;
 use Shlinkio\Shlink\CLI\Util\GeolocationDbUpdater;
 use Shlinkio\Shlink\IpGeolocation\Resolver\IpLocationResolverInterface;
+use Symfony\Component\Mercure\Publisher;
 
 return [
 
     'events' => [
         'regular' => [
             EventDispatcher\VisitLocated::class => [
+                EventDispatcher\NotifyVisitToMercure::class,
                 EventDispatcher\NotifyVisitToWebHooks::class,
             ],
         ],
@@ -28,6 +30,7 @@ return [
         'factories' => [
             EventDispatcher\LocateShortUrlVisit::class => ConfigAbstractFactory::class,
             EventDispatcher\NotifyVisitToWebHooks::class => ConfigAbstractFactory::class,
+            EventDispatcher\NotifyVisitToMercure::class => ConfigAbstractFactory::class,
         ],
 
         'delegators' => [
@@ -53,6 +56,12 @@ return [
             'config.url_shortener.domain',
             Options\AppOptions::class,
         ],
+        EventDispatcher\NotifyVisitToMercure::class => [
+            Publisher::class,
+            Mercure\MercureUpdatesGenerator::class,
+            'em',
+            'Logger_Shlink',
+        ],
     ],
 
 ];

module/Core/src/EventDispatcher/NotifyVisitToMercure.php

@@ -0,0 +1,54 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Shlinkio\Shlink\Core\EventDispatcher;
+
+use Doctrine\ORM\EntityManagerInterface;
+use Psr\Log\LoggerInterface;
+use Shlinkio\Shlink\Core\Entity\Visit;
+use Shlinkio\Shlink\Core\Mercure\MercureUpdatesGeneratorInterface;
+use Symfony\Component\Mercure\PublisherInterface;
+use Throwable;
+
+class NotifyVisitToMercure
+{
+    private PublisherInterface $publisher;
+    private MercureUpdatesGeneratorInterface $updatesGenerator;
+    private EntityManagerInterface $em;
+    private LoggerInterface $logger;
+
+    public function __construct(
+        PublisherInterface $publisher,
+        MercureUpdatesGeneratorInterface $updatesGenerator,
+        EntityManagerInterface $em,
+        LoggerInterface $logger
+    ) {
+        $this->publisher = $publisher;
+        $this->em = $em;
+        $this->logger = $logger;
+        $this->updatesGenerator = $updatesGenerator;
+    }
+
+    public function __invoke(VisitLocated $shortUrlLocated): void
+    {
+        $visitId = $shortUrlLocated->visitId();
+
+        /** @var Visit|null $visit */
+        $visit = $this->em->find(Visit::class, $visitId);
+        if ($visit === null) {
+            $this->logger->warning('Tried to notify mercure for visit with id "{visitId}", but it does not exist.', [
+                'visitId' => $visitId,
+            ]);
+            return;
+        }
+
+        try {
+            ($this->publisher)($this->updatesGenerator->newVisitUpdate($visit));
+        } catch (Throwable $e) {
+            $this->logger->debug('Error while trying to notify mercure hub with new visit. {e}', [
+                'e' => $e,
+            ]);
+        }
+    }
+}

module/Core/src/Mercure/MercureUpdatesGenerator.php

@@ -0,0 +1,33 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Shlinkio\Shlink\Core\Mercure;
+
+use Shlinkio\Shlink\Core\Entity\Visit;
+use Shlinkio\Shlink\Core\Transformer\ShortUrlDataTransformer;
+use Symfony\Component\Mercure\Update;
+
+use function json_encode;
+
+use const JSON_THROW_ON_ERROR;
+
+final class MercureUpdatesGenerator implements MercureUpdatesGeneratorInterface
+{
+    private const NEW_VISIT_TOPIC = 'https://shlink.io/new_visit';
+
+    private ShortUrlDataTransformer $transformer;
+
+    public function __construct(array $domainConfig)
+    {
+        $this->transformer = new ShortUrlDataTransformer($domainConfig);
+    }
+
+    public function newVisitUpdate(Visit $visit): Update
+    {
+        return new Update(self::NEW_VISIT_TOPIC, json_encode([
+            'shortUrl' => $this->transformer->transform($visit->getShortUrl()),
+            'visit' => $visit->jsonSerialize(),
+        ], JSON_THROW_ON_ERROR));
+    }
+}

module/Core/src/Mercure/MercureUpdatesGeneratorInterface.php

@@ -0,0 +1,13 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Shlinkio\Shlink\Core\Mercure;
+
+use Shlinkio\Shlink\Core\Entity\Visit;
+use Symfony\Component\Mercure\Update;
+
+interface MercureUpdatesGeneratorInterface
+{
+    public function newVisitUpdate(Visit $visit): Update;
+}