Tidies Pusher endpoints

composer.json

@@ -19,11 +19,7 @@
         "aws/aws-sdk-php": "^3.241",
         "clue/redis-react": "^2.6",
         "guzzlehttp/psr7": "^2.6",
-        "illuminate/cache": "^10.0",
-        "illuminate/console": "^10.0",
-        "illuminate/http": "^10.0",
-        "illuminate/redis": "^10.0",
-        "illuminate/support": "^10.0",
+        "illuminate/contracts": "^10.0",
         "ratchet/rfc6455": "^0.3.1",
         "react/socket": "^1.14",
         "symfony/http-foundation": "^6.3"

src/Http/Request.php

@@ -20,7 +20,7 @@ class Request
      *
      * @var int
      */
-    const MAX_SIZE = 4096;
+    const MAX_SIZE = 10000;
 
     /**
      * Turn the raw message into a Psr7 request.

src/Pusher/Concerns/InteractsWithChannelInformation.php

@@ -0,0 +1,65 @@
+<?php
+
+namespace Laravel\Reverb\Pusher\Concerns;
+
+use Laravel\Reverb\Channels\CacheChannel;
+use Laravel\Reverb\Channels\Channel;
+use Laravel\Reverb\Channels\Concerns\InteractsWithPresenceChannels;
+use Laravel\Reverb\Contracts\ChannelManager;
+
+trait InteractsWithChannelInformation
+{
+    /**
+     * Get the info for the given channels.
+     */
+    protected function infoForChannels(array $channels, string $info): array
+    {
+        return collect($channels)->mapWithKeys(function ($channel) use ($info) {
+            $name = $channel instanceof Channel ? $channel->name() : $channel;
+
+            return [$name => $this->info($name, $info)];
+        })->all();
+    }
+
+    /**
+     * Get the info for the given channels.
+     *
+     * @param  array<int, string>  $channels
+     * @return array<string, array<string, int>>
+     */
+    protected function info(string $channel, string $info): array
+    {
+        $info = explode(',', $info);
+
+        if (! $channel = app(ChannelManager::class)->find($channel)) {
+            return [];
+        }
+
+        $count = count($channel->connections());
+
+        $info = [
+            'occupied' => in_array('occupied', $info) ? $count > 0 : null,
+            'user_count' => in_array('user_count', $info) && $this->isPresenceChannel($channel) ? $count : null,
+            'subscription_count' => in_array('subscription_count', $info) && ! $this->isPresenceChannel($channel) ? $count : null,
+            'cache' => in_array('cache', $info) && $this->isCacheChannel($channel) ? $channel->cachedPayload() : null,
+        ];
+
+        return array_filter($info, fn ($item) => $item !== null);
+    }
+
+    /**
+     * Determine if the channel is a presence channel.
+     */
+    protected function isPresenceChannel(Channel $channel): bool
+    {
+        return in_array(InteractsWithPresenceChannels::class, class_uses($channel));
+    }
+
+    /**
+     * Determine if the channel is a cache channel.
+     */
+    protected function isCacheChannel(Channel $channel): bool
+    {
+        return $channel instanceof CacheChannel;
+    }
+}

src/Pusher/Http/Controllers/ChannelController.php

@@ -3,28 +3,22 @@
 namespace Laravel\Reverb\Pusher\Http\Controllers;
 
 use Laravel\Reverb\Http\Connection;
+use Laravel\Reverb\Pusher\Concerns\InteractsWithChannelInformation;
 use Psr\Http\Message\RequestInterface;
 use Symfony\Component\HttpFoundation\JsonResponse;
 use Symfony\Component\HttpFoundation\Response;
 
 class ChannelController extends Controller
 {
+    use InteractsWithChannelInformation;
+
     /**
      * Handle the request.
      */
     public function __invoke(RequestInterface $request, Connection $connection, string $appId, string $channel): Response
     {
         $this->verify($request, $connection, $appId);
 
-        $info = explode(',', $this->query['info'] ?? '');
-        $connections = $this->channels->find($channel)->connections();
-        $totalConnections = count($connections);
-
-        return new JsonResponse((object) array_filter([
-            'occupied' => $totalConnections > 0,
-            'user_count' => in_array('user_count', $info) ? $totalConnections : null,
-            'subscription_count' => in_array('subscription_count', $info) ? $totalConnections : null,
-            'cache' => in_array('cache', $info) ? '{}' : null,
-        ], fn ($item) => $item !== null));
+        return new JsonResponse((object) $this->info($channel, ($this->query['info'] ?? '').',occupied'));
     }
 }

src/Pusher/Http/Controllers/ChannelUsersController.php

@@ -2,14 +2,16 @@
 
 namespace Laravel\Reverb\Pusher\Http\Controllers;
 
-use Laravel\Reverb\Channels\PresenceChannel;
 use Laravel\Reverb\Http\Connection;
+use Laravel\Reverb\Pusher\Concerns\InteractsWithChannelInformation;
 use Psr\Http\Message\RequestInterface;
 use Symfony\Component\HttpFoundation\JsonResponse;
 use Symfony\Component\HttpFoundation\Response;
 
 class ChannelUsersController extends Controller
 {
+    use InteractsWithChannelInformation;
+
     /**
      * Handle the request.
      */
@@ -19,7 +21,7 @@ public function __invoke(RequestInterface $request, Connection $connection, stri
 
         $channel = $this->channels->find($channel);
 
-        if (! $channel instanceof PresenceChannel) {
+        if (! $this->isPresenceChannel($channel)) {
             return new JsonResponse((object) [], 400);
         }
 

src/Pusher/Http/Controllers/ChannelsController.php

@@ -4,12 +4,15 @@
 
 use Illuminate\Support\Str;
 use Laravel\Reverb\Http\Connection;
+use Laravel\Reverb\Pusher\Concerns\InteractsWithChannelInformation;
 use Psr\Http\Message\RequestInterface;
 use Symfony\Component\HttpFoundation\JsonResponse;
 use Symfony\Component\HttpFoundation\Response;
 
 class ChannelsController extends Controller
 {
+    use InteractsWithChannelInformation;
+
     /**
      * Handle the request.
      */
@@ -18,16 +21,20 @@ public function __invoke(RequestInterface $request, Connection $connection, stri
         $this->verify($request, $connection, $appId);
 
         $channels = collect($this->channels->all());
-        $info = explode(',', $this->query['info'] ?? '');
 
         if (isset($this->query['filter_by_prefix'])) {
             $channels = $channels->filter(fn ($channel) => Str::startsWith($channel->name(), $this->query['filter_by_prefix']));
         }
 
-        $channels = $channels->mapWithKeys(function ($channel) use ($info) {
-            return [$channel->name() => array_filter(['user_count' => in_array('user_count', $info) ? count($channel->connections()) : null])];
-        });
+        $channels = $channels->filter(fn ($channel) => count($channel->connections()) > 0);
+
+        $channels = $this->infoForChannels(
+            $channels->all(),
+            $this->query['info'] ?? ''
+        );
 
-        return new JsonResponse((object) ['channels' => $channels]);
+        return new JsonResponse([
+            'channels' => array_map(fn ($item) => (object) $item, $channels),
+        ]);
     }
 }

src/Pusher/Http/Controllers/EventsBatchController.php

@@ -2,23 +2,35 @@
 
 namespace Laravel\Reverb\Pusher\Http\Controllers;
 
+use Illuminate\Contracts\Validation\Validator;
+use Illuminate\Support\Facades\Validator as ValidatorFacade;
 use Laravel\Reverb\Event;
 use Laravel\Reverb\Http\Connection;
+use Laravel\Reverb\Pusher\Concerns\InteractsWithChannelInformation;
 use Psr\Http\Message\RequestInterface;
 use Symfony\Component\HttpFoundation\JsonResponse;
 use Symfony\Component\HttpFoundation\Response;
 
 class EventsBatchController extends Controller
 {
+    use InteractsWithChannelInformation;
+
     /**
      * Handle the request.
      */
     public function __invoke(RequestInterface $request, Connection $connection, string $appId): Response
     {
         $this->verify($request, $connection, $appId);
-        // @TODO Validate the request body as a JSON array of events in the correct format and a max of 10 items.
 
-        $items = collect(json_decode($this->body, true));
+        $payload = json_decode($this->body, true);
+
+        $validator = $this->validate($payload);
+
+        if ($validator->fails()) {
+            return new JsonResponse($validator->errors(), 422);
+        }
+
+        $items = collect($payload['batch']);
 
         $info = $items->map(function ($item) {
             Event::dispatch(
@@ -31,10 +43,16 @@ public function __invoke(RequestInterface $request, Connection $connection, stri
                 isset($item['socket_id']) ? ($this->channels->connections()[$item['socket_id']] ?? null) : null
             );
 
-            return isset($item['info']) ? $this->getInfo($item['channel'], $item['info']) : [];
+            return isset($item['info']) ? $this->info($item['channel'], $item['info']) : [];
         });
 
-        return $info->some(fn ($item) => count($item) > 0) ? new JsonResponse((object) ['batch' => $info->all()]) : new JsonResponse((object) []);
+        if ($info->some(fn ($item) => count($item) > 0)) {
+            return new JsonResponse(
+                ['batch' => $info->each(fn ($item) => (object) $item)->all()]
+            );
+        }
+
+        return new JsonResponse(['batch' => (object) []]);
     }
 
     /**
@@ -54,4 +72,19 @@ protected function getInfo(string $channel, string $info): array
 
         return array_filter($info, fn ($item) => $item !== null);
     }
+
+    /**
+     * Validate the incoming request.
+     */
+    protected function validate(array $payload): Validator
+    {
+        return ValidatorFacade::make($payload, [
+            'batch' => ['required', 'array'],
+            'batch.*.name' => ['required', 'string'],
+            'batch.*.data' => ['required', 'array'],
+            'batch.*.channel' => ['required_without:channels', 'string'],
+            'batch.*.socket_id' => ['string'],
+            'batch.*.info' => ['string'],
+        ]);
+    }
 }

src/Pusher/Http/Controllers/EventsController.php

@@ -2,24 +2,35 @@
 
 namespace Laravel\Reverb\Pusher\Http\Controllers;
 
+use Illuminate\Contracts\Validation\Validator;
 use Illuminate\Support\Arr;
+use Illuminate\Support\Facades\Validator as ValidatorFacade;
 use Laravel\Reverb\Event;
 use Laravel\Reverb\Http\Connection;
+use Laravel\Reverb\Pusher\Concerns\InteractsWithChannelInformation;
 use Psr\Http\Message\RequestInterface;
 use Symfony\Component\HttpFoundation\JsonResponse;
 use Symfony\Component\HttpFoundation\Response;
 
 class EventsController extends Controller
 {
+    use InteractsWithChannelInformation;
+
     /**
      * Handle the request.
      */
     public function __invoke(RequestInterface $request, Connection $connection, string $appId): Response
     {
         $this->verify($request, $connection, $appId);
-        // @TODO Validate the request body as a JSON object in the correct format.
 
         $payload = json_decode($this->body, true);
+
+        $validator = $this->validate($payload);
+
+        if ($validator->fails()) {
+            return new JsonResponse($validator->errors(), 422);
+        }
+
         $channels = Arr::wrap($payload['channels'] ?? $payload['channel'] ?? []);
 
         Event::dispatch(
@@ -33,32 +44,26 @@ public function __invoke(RequestInterface $request, Connection $connection, stri
         );
 
         if (isset($payload['info'])) {
-            return new JsonResponse((object) $this->getInfo($channels, $payload['info']));
+            return new JsonResponse([
+                'channels' => array_map(fn ($item) => (object) $item, $this->infoForChannels($channels, $payload['info'])),
+            ]);
         }
 
         return new JsonResponse((object) []);
     }
 
     /**
-     * Get the info for the given channels.
-     *
-     * @param  array<int, string>  $channels
-     * @return array<string, array<string, int>>
+     * Validate the incoming request.
      */
-    protected function getInfo(array $channels, string $info): array
+    protected function validate(array $payload): Validator
     {
-        $info = explode(',', $info);
-
-        $channels = collect($channels)->mapWithKeys(function ($channel) use ($info) {
-            $count = count($this->channels->find($channel)->connections());
-            $info = [
-                'user_count' => in_array('user_count', $info) ? $count : null,
-                'subscription_count' => in_array('subscription_count', $info) ? $count : null,
-            ];
-
-            return [$channel => array_filter($info, fn ($item) => $item !== null)];
-        })->all();
-
-        return ['channels' => $channels];
+        return ValidatorFacade::make($payload, [
+            'name' => ['required', 'string'],
+            'data' => ['required', 'array'],
+            'channels' => ['required_without:channel', 'array'],
+            'channel' => ['required_without:channels', 'string'],
+            'socket_id' => ['string'],
+            'info' => ['string'],
+        ]);
     }
 }

src/Pusher/Http/Controllers/UsersTerminateController.php

@@ -16,11 +16,13 @@ public function __invoke(RequestInterface $request, Connection $connection, stri
     {
         $this->verify($request, $connection, $appId);
 
-        if (! $connection = $this->channels->connections()[$userId]) {
-            return new JsonResponse((object) [], 400);
-        }
+        $connections = collect($this->channels->connections());
 
-        $connection->connection()->disconnect();
+        $connections->each(function ($connection) use ($userId) {
+            if ((string) $connection->data()['user_id'] === $userId) {
+                $connection->disconnect();
+            }
+        });
 
         return new JsonResponse((object) []);
     }