ISSUE #1675: Postpone webhooks process when other process is importing data

Author: robiningelbrecht

composer.lock

@@ -1098,6 +1098,17 @@
  *     doctrine?: DoctrineConfig,
  *     doctrine_migrations?: DoctrineMigrationsConfig,
  *     twig?: TwigConfig,
+ *     "when@dev"?: array{
+ *         imports?: ImportsConfig,
+ *         parameters?: ParametersConfig,
+ *         services?: ServicesConfig,
+ *         framework?: FrameworkConfig,
+ *         flysystem?: FlysystemConfig,
+ *         monolog?: MonologConfig,
+ *         doctrine?: DoctrineConfig,
+ *         doctrine_migrations?: DoctrineMigrationsConfig,
+ *         twig?: TwigConfig,
+ *     },
  *     "when@prod"?: array{
  *         imports?: ImportsConfig,
  *         parameters?: ParametersConfig,
@@ -1200,6 +1211,7 @@ public static function config(array $config): array
  *     deprecated?: array{package:string, version:string, message?:string},
  * }
  * @psalm-type RoutesConfig = array{
+ *     "when@dev"?: array<string, RouteConfig|ImportConfig|AliasConfig>,
  *     "when@prod"?: array<string, RouteConfig|ImportConfig|AliasConfig>,
  *     "when@test"?: array<string, RouteConfig|ImportConfig|AliasConfig>,
  *     ...<string, RouteConfig|ImportConfig|AliasConfig>

config/reference.php

@@ -14,6 +14,7 @@
 use App\Domain\Strava\Webhook\WebhookEventRepository;
 use App\Infrastructure\CQRS\Command\Bus\CommandBus;
 use App\Infrastructure\Daemon\Cron\RunnableCronAction;
+use App\Infrastructure\Daemon\Mutex\LockIsAlreadyAcquired;
 use App\Infrastructure\Daemon\Mutex\LockName;
 use App\Infrastructure\Daemon\Mutex\Mutex;
 use App\Infrastructure\DependencyInjection\Mutex\WithMutex;
@@ -66,7 +67,12 @@ public function run(SymfonyStyle $output): void
     public function runForWebhooks(SymfonyStyle $output): void
     {
         $this->migrationRunner->run($output);
-        $this->mutex->acquireLock('importDataAndBuildAppCronAction');
+        try {
+            $this->mutex->acquireLock('processWebhooks');
+        } catch (LockIsAlreadyAcquired) {
+            // Another process is importing data, postpone webhook processing.
+            return;
+        }
 
         if (!$webhookEvents = $this->webhookEventRepository->grab()) {
             // No webhooks to process.

src/Application/importDataAndBuildAppCronAction.php

@@ -52,7 +52,7 @@ public function __construct(
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
         $output = new SymfonyStyle($input, $output);
-        $this->mutex->acquireLock('BuildAppConsoleCommand');
+        $this->mutex->acquireLock('DetectCorruptedActivitiesConsoleCommand');
 
         $this->outputConsoleIntro($output);
 

src/Console/DetectCorruptedActivitiesConsoleCommand.php

@@ -49,7 +49,7 @@ public function acquireLock(string $lockAcquiredBy): void
         }
 
         $data = Json::decode($row);
-        $heartbeat = $data[self::HEARTBEAT_KEY];
+        $heartbeat = $data[self::HEARTBEAT_KEY] ?? null;
         $isStale = $now - $heartbeat > self::STALE_THRESHOLD_IN_SECONDS;
 
         if ($isStale) {

src/Infrastructure/Daemon/Mutex/Mutex.php

@@ -82,6 +82,31 @@ public function testRunForWebhooks(): void
         $this->assertMatchesJsonSnapshot(Json::encode($this->commandBus->getDispatchedCommands()));
     }
 
+    public function testRunForWebhooksWhenLockIsAlreadyAcquired(): void
+    {
+        $output = new SpySymfonyStyleOutput(new StringInput('input'), new NullOutput());
+
+        $event = WebhookEvent::create(
+            objectId: '1',
+            objectType: 'activity',
+            aspectType: WebhookAspectType::CREATE,
+            payload: [],
+        );
+        $this->getContainer()->get(WebhookEventRepository::class)->add($event);
+
+        $this->getConnection()->executeStatement(
+            'INSERT INTO KeyValue (`key`, `value`) VALUES (:key, :value)',
+            ['key' => 'lock.importDataOrBuildApp', 'value' => '{"lockAcquiredBy": "test", "heartbeat": 1764806400}']
+        );
+
+        $this->migrationRunner
+            ->expects($this->once())
+            ->method('run');
+
+        $this->importAndBuildAppCronAction->runForWebhooks($output);
+        $this->assertEmpty($this->commandBus->getDispatchedCommands());
+    }
+
     #[\Override]
     protected function setUp(): void
     {