working (but messy) lottery queue with mercure example

Author: clegginabox

Caddyfile.dev

@@ -1,5 +1,6 @@
 :80 {
   root * {$SERVER_ROOT}
   file_server
+  reverse_proxy /.well-known/mercure* mercure:80
   php_server
 }

README.md

@@ -2,9 +2,6 @@
 
 [![Codacy Badge](https://app.codacy.com/project/badge/Grade/cdc12dbceac04dc8bbece4012222cd3d)](https://app.codacy.com/gh/clegginabox/airlock-php/dashboard?utm_source=gh&utm_medium=referral&utm_content=&utm_campaign=Badge_grade)
 [![Codacy Badge](https://app.codacy.com/project/badge/Coverage/cdc12dbceac04dc8bbece4012222cd3d)](https://app.codacy.com/gh/clegginabox/airlock-php/dashboard?utm_source=gh&utm_medium=referral&utm_content=&utm_campaign=Badge_coverage)
-![PHPCS](https://img.shields.io/github/actions/workflow/status/clegginabox/airlock-php/tests.yaml?label=phpcs)
-![PHPUnit](https://img.shields.io/github/actions/workflow/status/clegginabox/airlock-php/tests.yaml?label=tests)
-![E2E](https://img.shields.io/github/actions/workflow/status/clegginabox/airlock-php/tests.yaml?label=e2e)
 ![PHPStan Level 8](https://img.shields.io/badge/phpstan%20level-8%20of%209-green?style=flat-square&logo=php)
 
 [![Quality Gate Status](https://sonarcloud.io/api/project_badges/measure?project=clegginabox_airlock-php&metric=alert_status)](https://sonarcloud.io/summary/new_code?id=clegginabox_airlock-php)

compose.yaml

@@ -13,29 +13,30 @@ services:
       REDIS_HOST: redis
       REDIS_PORT: 6379
       SERVER_ROOT: /app/examples/public
+      MERCURE_HUB_URL: http://mercure/.well-known/mercure
+      MERCURE_JWT_SECRET: ${MERCURE_JWT_SECRET:-airlock-mercure-secret-32chars-minimum}
+      MERCURE_PUBLIC_URL: ${MERCURE_PUBLIC_URL:-http://localhost/.well-known/mercure}
     env_file:
       - .env
     tty: true
     depends_on:
       - redis
+      - mercure
     ports:
       - "80:80" # HTTP
 
-  worker:
-    build:
-      context: .
-      dockerfile: Dockerfile
-    working_dir: /app
-    volumes:
-      - ./:/app
+  mercure:
+    image: dunglas/mercure
+    ports:
+      - "3000:80"
     environment:
-      REDIS_HOST: redis
-      REDIS_PORT: 6379
-    env_file:
-      - .env
-    depends_on:
-      - redis
-    command: php examples/bin/console app:worker
+      SERVER_NAME: :80
+      MERCURE_PUBLISHER_JWT_KEY: ${MERCURE_JWT_SECRET:-airlock-mercure-secret-32chars-minimum}
+      MERCURE_SUBSCRIBER_JWT_KEY: ${MERCURE_JWT_SECRET:-airlock-mercure-secret-32chars-minimum}
+      MERCURE_PUBLISHER_JWT_ALG: HS256
+      MERCURE_SUBSCRIBER_JWT_ALG: HS256
+      MERCURE_ALLOW_ANONYMOUS: 1
+      MERCURE_CORS_ALLOWED_ORIGINS: "*"
 
   redis:
     image: redis:8.4.0
@@ -54,7 +55,6 @@ services:
     env_file:
       - .env
     depends_on:
-      - worker
       - redis
     command: >
       sh -c "npm install && npx playwright test"

examples/bin/console

@@ -6,7 +6,7 @@ declare(strict_types=1);
 use App\Kernel;
 use Symfony\Bundle\FrameworkBundle\Console\Application;
 
-require dirname(__DIR__) . '/vendor/autoload.php';
+require_once dirname(__DIR__) . '/vendor/autoload.php';
 
 $env = $_SERVER['APP_ENV'] ?? 'dev';
 $debug = (bool) ($_SERVER['APP_DEBUG'] ?? true);

examples/composer.json

@@ -23,7 +23,9 @@
         "symfony/semaphore": "^8.0",
         "symfony/console": "^8.0",
         "symfony/dotenv": "^8.0",
-        "symfony/yaml": "^8.0"
+        "symfony/yaml": "^8.0",
+        "symfony/mercure": "^0.7.2",
+        "lcobucci/jwt": "^5.6"
     },
     "repositories": [
         {

examples/config/framework.yaml

@@ -1,2 +1,4 @@
 framework:
   secret: S0ME_SECRET
+  session:
+    enabled: true

examples/src/Command/WorkerCommand.php

@@ -6,6 +6,8 @@
 
 use App\GlobalLock\GlobalLock;
 use App\RedisLotteryQueue\RedisLotteryQueue;
+use Clegginabox\Airlock\Bridge\Mercure\MercureAirlockNotifier;
+use Clegginabox\Airlock\Bridge\Symfony\Mercure\SymfonyMercureHubFactory;
 use Clegginabox\Airlock\Bridge\Symfony\Seal\SymfonyLockSeal;
 use Clegginabox\Airlock\Bridge\Symfony\Seal\SymfonySemaphoreSeal;
 use Clegginabox\Airlock\Notifier\NullAirlockNotifier;
@@ -81,4 +83,34 @@ public function redisLotteryQueue(
 
         return new QueueAirlock($seal, $queue, new NullAirlockNotifier());
     }
+
+    public function redisLotteryQueueWithMercure(
+        int $limit = 3,
+        int $ttl = 60,
+        int $claimWindow = 10,
+    ): QueueAirlock {
+        $seal = new SymfonySemaphoreSeal(
+            factory: new SemaphoreFactory(new SemaphoreRedisStore($this->redis)),
+            resource: RedisLotteryQueue::RESOURCE->value,
+            limit: $limit,
+            weight: 1,
+            ttlInSeconds: $ttl,
+            autoRelease: false,
+        );
+
+        $queue = new LotteryQueue(
+            new RedisLotteryQueueStore(
+                redis: $this->redis,
+                setKey: RedisLotteryQueue::SET_KEY->value,
+                candidateKey: RedisLotteryQueue::CANDIDATE_KEY->value,
+                candidateTtlSeconds: $claimWindow,
+            ),
+        );
+
+        $hubUrl = getenv('MERCURE_HUB_URL') ?: 'http://localhost:3000/.well-known/mercure';
+        $jwtSecret = getenv('MERCURE_JWT_SECRET') ?: 'airlock-mercure-secret';
+        $hub = SymfonyMercureHubFactory::create($hubUrl, $jwtSecret);
+
+        return new QueueAirlock($seal, $queue, new MercureAirlockNotifier($hub));
+    }
 }

examples/src/Factory/AirlockFactory.php

@@ -6,13 +6,16 @@
 
 use App\Infrastructure\ClientIdCookieSubscriber;
 use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
+use Symfony\Component\HttpFoundation\Exception\SessionNotFoundException;
 use Symfony\Component\HttpFoundation\JsonResponse;
 use Symfony\Component\HttpFoundation\Request;
 use Symfony\Component\HttpFoundation\Response;
 use Symfony\Component\Routing\Attribute\Route;
 
 class Controller extends AbstractController
 {
+    private const SESSION_TOKEN_KEY = 'airlock.redis_lottery.token';
+
     #[Route('/redis-lottery-queue', methods: [Request::METHOD_GET])]
     public function index(): Response
     {
@@ -39,17 +42,51 @@ public function success(): Response
         );
     }
 
+    #[Route('/redis-lottery-queue/release', methods: [Request::METHOD_POST])]
+    public function release(Request $request, RedisLotteryQueueService $service): Response
+    {
+        try {
+            $session = $request->getSession();
+        } catch (SessionNotFoundException) {
+            return new JsonResponse(['ok' => false, 'error' => 'Session not available'], Response::HTTP_BAD_REQUEST);
+        }
+
+        $serializedToken = $session->get(self::SESSION_TOKEN_KEY);
+        if (!is_string($serializedToken) || $serializedToken === '') {
+            return new JsonResponse(['ok' => false, 'error' => 'Missing airlock token'], Response::HTTP_BAD_REQUEST);
+        }
+
+        $service->release($serializedToken);
+        $session->remove(self::SESSION_TOKEN_KEY);
+
+        return new JsonResponse(['ok' => true]);
+    }
+
     #[Route('/redis-lottery-queue/start', methods: [Request::METHOD_POST])]
     public function start(Request $request, RedisLotteryQueueService $service): JsonResponse
     {
         $clientId = (string) $request->attributes->get(ClientIdCookieSubscriber::ATTRIBUTE);
+
         $result = $service->start($clientId);
 
         if ($result->isAdmitted()) {
+            try {
+                $session = $request->getSession();
+                $token = $result->getToken();
+                if ($token !== null) {
+                    $session->set(self::SESSION_TOKEN_KEY, (string) $token);
+                }
+            } catch (SessionNotFoundException) {
+                // Session not available; continue without persisting token.
+            }
+
             return new JsonResponse([
                 'ok' => true,
                 'status' => 'admitted',
                 'clientId' => $clientId,
+                'topic' => $service->getTopic($clientId),
+                'hubUrl' => $service->getHubUrl(),
+                'token' => $service->getSubscriberToken($clientId),
             ]);
         }
 
@@ -58,6 +95,9 @@ public function start(Request $request, RedisLotteryQueueService $service): Json
             'status' => 'queued',
             'position' => $result->getPosition(),
             'clientId' => $clientId,
+            'topic' => $service->getTopic($clientId),
+            'hubUrl' => $service->getHubUrl(),
+            'token' => $service->getSubscriberToken($clientId),
         ]);
     }
 }

examples/src/RedisLotteryQueue/Controller.php

@@ -5,8 +5,11 @@
 namespace App\RedisLotteryQueue;
 
 use App\Factory\AirlockFactory;
+use Clegginabox\Airlock\Bridge\Symfony\Seal\SymfonySemaphoreToken;
 use Clegginabox\Airlock\EntryResult;
 use Clegginabox\Airlock\QueueAirlock;
+use Symfony\Component\Mercure\Jwt\LcobucciFactory;
+use Symfony\Component\Semaphore\Key;
 
 final class RedisLotteryQueueService
 {
@@ -15,27 +18,43 @@ final class RedisLotteryQueueService
     public function __construct(
         private readonly AirlockFactory $airlockFactory,
     ) {
-        $this->airlock = $this->airlockFactory->redisLotteryQueue();
+        $this->airlock = $this->airlockFactory->redisLotteryQueueWithMercure(
+            limit: 1, // Only allow one person "inside" at a time
+            ttl: 60, // 60 second time limit
+            claimWindow: 10, // 10 seconds to claim the place
+        );
     }
 
-    public function start(string $clientId, int $holdSeconds = 15): EntryResult
+    public function start(string $clientId): EntryResult
     {
-        return $this->tryAdmit($clientId, $holdSeconds);
+        return $this->airlock->enter($clientId);
     }
 
-    public function check(string $clientId, int $holdSeconds = 15): EntryResult
+    public function release(string $serializedToken): void
     {
-        return $this->tryAdmit($clientId, $holdSeconds);
+        $key = unserialize($serializedToken, ['allowed_classes' => [Key::class]]);
+        if (!$key instanceof Key) {
+            return;
+        }
+
+        $this->airlock->release(new SymfonySemaphoreToken($key));
     }
 
-    private function tryAdmit(string $clientId, int $holdSeconds = 15): EntryResult
+    public function getTopic(string $clientId): string
     {
-        $result = $this->airlock->enter($clientId);
+        return $this->airlock->getTopic($clientId);
+    }
 
-        if ($result->isAdmitted()) {
-            return $result;
-        }
+    public function getHubUrl(): string
+    {
+        return getenv('MERCURE_PUBLIC_URL') ?: 'http://localhost:3000/.well-known/mercure';
+    }
+
+    public function getSubscriberToken(string $clientId): string
+    {
+        $jwtSecret = getenv('MERCURE_JWT_SECRET') ?: 'airlock-mercure-secret-32chars-minimum';
 
-        return $result;
+        return new LcobucciFactory($jwtSecret)
+            ->create(subscribe: [$this->getTopic($clientId)]);
     }
 }