refactor: separate event queue mainsail

app/Events/Concerns/ShouldBeUniqueEvent.php

@@ -10,29 +10,35 @@ trait ShouldBeUniqueEvent
 {
     public const UNIQUE_KEY = 'webhooks:event';
 
+    protected array $ids = [null];
+
+    protected ?array $broadcastChannels = null;
+
     final public function broadcastWhen(): bool
     {
-        $lock = Cache::lock(
-            $this->uniqueKey(),
-            $this->uniqueTimeout()
-        );
+        $channels = [];
+
+        foreach ($this->ids as $id) {
+            $channelName = $this->channelName($id);
+            $lock        = Cache::lock($this->uniqueKeyForChannel($channelName), $this->uniqueTimeout());
 
-        if ($lock->get() === false) {
-            return false;
+            /** @var bool $hasAcquiredLock */
+            $hasAcquiredLock = $lock->get();
+
+            if ($hasAcquiredLock) {
+                $channels[] = $channelName;
+            }
         }
 
-        return true;
+        $this->broadcastChannels = $channels;
+
+        return count($channels) > 0;
     }
 
     abstract protected function uniqueTimeout(): int;
 
-    // @phpstan-ignore-next-line
-    protected function uniqueKey(): string
+    protected function uniqueKeyForChannel(string $channelName): string
     {
-        return sprintf(
-            '%s:%s',
-            static::UNIQUE_KEY,
-            $this->channelName()
-        );
+        return sprintf('%s:%s', static::UNIQUE_KEY, $channelName);
     }
 }

app/Events/Statistics/StatisticsUpdate.php

@@ -15,12 +15,12 @@ final protected function uniqueTimeout(): int
         return config('arkscan.webhooks.statistics-update.ttl', 8);
     }
 
-    final protected function uniqueKey(): string
+    final protected function uniqueKeyForChannel(string $channelName): string
     {
         return sprintf(
             '%s:%s:%s',
             static::UNIQUE_KEY,
-            $this->channelName(),
+            $channelName,
             $this::class,
         );
     }

app/Events/WebsocketEvent.php

@@ -5,39 +5,46 @@
 namespace App\Events;
 
 use App\Events\Concerns\ShouldBeUniqueEvent;
-use Illuminate\Broadcasting\Channel;
+use Illuminate\Bus\Queueable;
 use Illuminate\Contracts\Broadcasting\ShouldBroadcast;
 use Illuminate\Foundation\Events\Dispatchable;
 
 abstract class WebsocketEvent implements ShouldBroadcast
 {
     use Dispatchable;
     use ShouldBeUniqueEvent;
+    use Queueable;
 
     public const CHANNEL = 'channel';
 
-    public function __construct(protected ?string $id = null)
+    public function __construct(string|null ...$ids)
     {
-        //
+        $this->ids = count($ids) === 0 ? [null] : array_values($ids);
+        $this->onQueue('reverb');
     }
 
-    final public function broadcastOn(): Channel
+    /** @return array<string> */
+    final public function broadcastOn(): array
     {
-        return new Channel($this->channelName());
+        // broadcastWhen() populates this after acquiring per-channel locks.
+        // When it hasn't been called (e.g. Event::fake() in tests), fall back
+        // to all channels so assertions against the event object still work.
+        return $this->broadcastChannels
+            ?? array_map(fn ($id) => $this->channelName($id), $this->ids);
     }
 
     final public function getId(): ?string
     {
-        return $this->id;
+        return $this->ids[0] ?? null;
     }
 
-    final protected function channelName(): string
+    final protected function channelName(?string $id = null): string
     {
-        if ($this->id !== null) {
+        if ($id !== null) {
             return sprintf(
                 '%s.%s',
                 static::CHANNEL,
-                $this->id
+                $id
             );
         }
 

app/Http/Controllers/Concerns/HandlesBlockWebhooks.php

@@ -11,14 +11,9 @@ trait HandlesBlockWebhooks
 {
     private function handleBlockApplied(): void
     {
-        NewBlock::dispatch();
+        NewBlock::dispatch(null, request()->input('data.proposer'));
 
         // We'll run the job instead of duplicating the logic as this is the only purpose for the job.
         CacheBlocks::dispatch();
     }
-
-    private function handleGeneratorBlockApplied(): void
-    {
-        NewBlock::dispatch(request()->input('data.proposer'));
-    }
 }

app/Http/Controllers/Concerns/HandlesTransactionWebhooks.php

@@ -12,18 +12,13 @@ trait HandlesTransactionWebhooks
 {
     private function handleTransactionApplied(): void
     {
-        NewTransaction::dispatch();
+        NewTransaction::dispatch(
+            null,
+            request()->input('data.senderPublicKey'),
+            request()->input('data.to'),
+        );
+
         CheckLatestWallet::dispatch();
         CheckLargestTransaction::dispatch();
     }
-
-    private function handleSenderTransactionApplied(): void
-    {
-        NewTransaction::dispatch(request()->input('data.senderPublicKey'));
-    }
-
-    private function handleRecipientTransactionApplied(): void
-    {
-        NewTransaction::dispatch(request()->input('data.to'));
-    }
 }

app/Http/Controllers/WebhooksController.php

@@ -24,15 +24,12 @@ public function __invoke(): void
 
         if ($event === WebhookEvents::BlockApplied->value) {
             $this->handleBlockApplied();
-            $this->handleGeneratorBlockApplied();
 
             return;
         }
 
         if ($event === WebhookEvents::TransactionApplied->value) {
             $this->handleTransactionApplied();
-            $this->handleSenderTransactionApplied();
-            $this->handleRecipientTransactionApplied();
 
             return;
         }

config/horizon.php

@@ -55,6 +55,16 @@
         'balanceCooldown' => 1,
         'tries'           => 1,
     ],
+    'reverb' => [
+        'connection'      => 'redis',
+        'queue'           => ['reverb'],
+        'balance'         => 'auto',
+        'minProcesses'    => 1,
+        'maxProcesses'    => 10,
+        'balanceMaxShift' => 5,
+        'balanceCooldown' => 1,
+        'tries'           => 1,
+    ],
 ];
 
 return [

tests/Feature/Events/NewBlockTest.php

@@ -11,7 +11,7 @@
     NewBlock::dispatch();
 
     Event::assertDispatched(NewBlock::class, function ($event) {
-        return $event->broadcastOn()->name === 'blocks';
+        return in_array('blocks', $event->broadcastOn(), true);
     });
 });
 
@@ -21,6 +21,6 @@
     NewBlock::dispatch('channel-id');
 
     Event::assertDispatched(NewBlock::class, function ($event) {
-        return $event->broadcastOn()->name === 'blocks.channel-id';
+        return in_array('blocks.channel-id', $event->broadcastOn(), true);
     });
 });

tests/Feature/Events/NewTransactionTest.php

@@ -11,7 +11,7 @@
     NewTransaction::dispatch();
 
     Event::assertDispatched(NewTransaction::class, function ($event) {
-        return $event->broadcastOn()->name === 'transactions';
+        return in_array('transactions', $event->broadcastOn(), true);
     });
 });
 
@@ -21,6 +21,6 @@
     NewTransaction::dispatch('channel-id');
 
     Event::assertDispatched(NewTransaction::class, function ($event) {
-        return $event->broadcastOn()->name === 'transactions.channel-id';
+        return in_array('transactions.channel-id', $event->broadcastOn(), true);
     });
 });