Merge branch '1.8.x' of https://github.com/appwrite/appwrite into joins2

app/cli.php

@@ -318,6 +318,10 @@ $setResource('logError', function (Registry $register) {
 
 $setResource('executor', fn () => new Executor(), []);
 
+$setResource('bus', function (Registry $register) use ($cli) {
+    return $register->get('bus')->setResolver(fn (string $name) => $cli->getResource($name));
+}, ['register']);
+
 $setResource('telemetry', fn () => new NoTelemetry(), []);
 
 $cli

app/config/collections/projects.php

@@ -2190,13 +2190,6 @@ return [
                 'lengths' => [],
                 'orders' => [Database::ORDER_ASC],
             ],
-            [
-                '$id' => ID::custom('_key_function_internal_id'),
-                'type' => Database::INDEX_KEY,
-                'attributes' => ['resourceInternalId'],
-                'lengths' => [],
-                'orders' => [],
-            ],
             [
                 '$id' => ID::custom('_key_resourceType'),
                 'type' => Database::INDEX_KEY,

app/controllers/general.php

@@ -5,11 +5,11 @@ require_once __DIR__ . '/../init.php';
 use Ahc\Jwt\JWT;
 use Ahc\Jwt\JWTException;
 use Appwrite\Auth\Key;
+use Appwrite\Bus\Events\ExecutionCompleted;
+use Appwrite\Bus\Events\RequestCompleted;
 use Appwrite\Event\Certificate;
 use Appwrite\Event\Delete as DeleteEvent;
 use Appwrite\Event\Event;
-use Appwrite\Event\Execution;
-use Appwrite\Event\StatsUsage;
 use Appwrite\Extend\Exception as AppwriteException;
 use Appwrite\Network\Cors;
 use Appwrite\Platform\Appwrite;
@@ -36,6 +36,7 @@ use Executor\Executor;
 use MaxMind\Db\Reader;
 use Swoole\Http\Request as SwooleRequest;
 use Swoole\Table;
+use Utopia\Bus\Bus;
 use Utopia\Config\Config;
 use Utopia\Console;
 use Utopia\Database\Database;
@@ -63,7 +64,7 @@ Config::setParam('domainVerification', false);
 Config::setParam('cookieDomain', 'localhost');
 Config::setParam('cookieSamesite', Response::COOKIE_SAMESITE_NONE);
 
-function router(Http $utopia, Database $dbForPlatform, callable $getProjectDB, SwooleRequest $swooleRequest, Request $request, Response $response, Log $log, Event $queueForEvents, StatsUsage $queueForStatsUsage, Execution $queueForExecutions, Executor $executor, Reader $geodb, callable $isResourceBlocked, array $platform, string $previewHostname, Authorization $authorization, ?Key $apiKey, DeleteEvent $queueForDeletes, int $executionsRetentionCount)
+function router(Http $utopia, Database $dbForPlatform, callable $getProjectDB, SwooleRequest $swooleRequest, Request $request, Response $response, Log $log, Event $queueForEvents, Bus $bus, Executor $executor, Reader $geodb, callable $isResourceBlocked, array $platform, string $previewHostname, Authorization $authorization, ?Key $apiKey, DeleteEvent $queueForDeletes, int $executionsRetentionCount)
 {
     $host = $request->getHostname() ?? '';
     if (!empty($previewHostname)) {
@@ -707,10 +708,12 @@ function router(Http $utopia, Database $dbForPlatform, callable $getProjectDB, S
             }
         } finally {
             if ($type === 'function' || $type === 'site') {
-                $queueForExecutions
-                    ->setExecution($execution)
-                    ->setProject($project)
-                    ->trigger();
+                $bus->dispatch(new ExecutionCompleted(
+                    execution: $execution->getArrayCopy(),
+                    project: $project->getArrayCopy(),
+                    spec: $spec,
+                    resource: $resource->getArrayCopy(),
+                ));
             }
         }
 
@@ -755,70 +758,12 @@ function router(Http $utopia, Database $dbForPlatform, callable $getProjectDB, S
             ->setStatusCode($execution['responseStatusCode'] ?? 200)
             ->send($body);
 
-        $fileSize = 0;
-        $file = $request->getFiles('file');
-        if (!empty($file)) {
-            $fileSize = (\is_array($file['size']) && isset($file['size'][0])) ? $file['size'][0] : $file['size'];
-        }
-
-        if (!empty($apiKey) && !empty($apiKey->getDisabledMetrics())) {
-            foreach ($apiKey->getDisabledMetrics() as $key) {
-                $queueForStatsUsage->disableMetric($key);
-            }
-        }
-
-        $metricTypeExecutions = str_replace(['{resourceType}'], [$deployment->getAttribute('resourceType')], METRIC_RESOURCE_TYPE_EXECUTIONS);
-        $metricTypeIdExecutions = str_replace(['{resourceType}', '{resourceInternalId}'], [$deployment->getAttribute('resourceType'), $resource->getSequence()], METRIC_RESOURCE_TYPE_ID_EXECUTIONS);
-        $metricTypeExecutionsCompute = str_replace(['{resourceType}'], [$deployment->getAttribute('resourceType')], METRIC_RESOURCE_TYPE_EXECUTIONS_COMPUTE);
-        $metricTypeIdExecutionsCompute = str_replace(['{resourceType}', '{resourceInternalId}'], [$deployment->getAttribute('resourceType'), $resource->getSequence()], METRIC_RESOURCE_TYPE_ID_EXECUTIONS_COMPUTE);
-        $metricTypeExecutionsMbSeconds = str_replace(['{resourceType}'], [$deployment->getAttribute('resourceType')], METRIC_RESOURCE_TYPE_EXECUTIONS_MB_SECONDS);
-        $metricTypeIdExecutionsMBSeconds = str_replace(['{resourceType}', '{resourceInternalId}'], [$deployment->getAttribute('resourceType'), $resource->getSequence()], METRIC_RESOURCE_TYPE_ID_EXECUTIONS_MB_SECONDS);
-        if ($deployment->getAttribute('resourceType') === 'sites') {
-            $queueForStatsUsage
-                ->disableMetric(METRIC_NETWORK_REQUESTS)
-                ->disableMetric(METRIC_NETWORK_INBOUND)
-                ->disableMetric(METRIC_NETWORK_OUTBOUND);
-            if ($resource->getAttribute('adapter') !== 'ssr') {
-                $queueForStatsUsage
-                    ->disableMetric(METRIC_EXECUTIONS)
-                    ->disableMetric(METRIC_EXECUTIONS_COMPUTE)
-                    ->disableMetric(METRIC_EXECUTIONS_MB_SECONDS)
-                    ->disableMetric($metricTypeExecutions)
-                    ->disableMetric($metricTypeIdExecutions)
-                    ->disableMetric($metricTypeExecutionsCompute)
-                    ->disableMetric($metricTypeIdExecutionsCompute)
-                    ->disableMetric($metricTypeExecutionsMbSeconds)
-                    ->disableMetric($metricTypeIdExecutionsMBSeconds);
-            }
-
-            $queueForStatsUsage
-                ->addMetric(METRIC_SITES_REQUESTS, 1)
-                ->addMetric(METRIC_SITES_INBOUND, $request->getSize() + $fileSize)
-                ->addMetric(METRIC_SITES_OUTBOUND, $response->getSize())
-                ->addMetric(str_replace('{siteInternalId}', $resource->getSequence(), METRIC_SITES_ID_REQUESTS), 1)
-                ->addMetric(str_replace('{siteInternalId}', $resource->getSequence(), METRIC_SITES_ID_INBOUND), $request->getSize() + $fileSize)
-                ->addMetric(str_replace('{siteInternalId}', $resource->getSequence(), METRIC_SITES_ID_OUTBOUND), $response->getSize())
-            ;
-        }
-
-
-        $compute = (int)($execution->getAttribute('duration') * 1000);
-        $mbSeconds = (int)(($spec['memory'] ?? APP_COMPUTE_MEMORY_DEFAULT) * $execution->getAttribute('duration', 0) * ($spec['cpus'] ?? APP_COMPUTE_CPUS_DEFAULT));
-        $queueForStatsUsage
-            ->addMetric(METRIC_NETWORK_REQUESTS, 1)
-            ->addMetric(METRIC_NETWORK_INBOUND, $request->getSize() + $fileSize)
-            ->addMetric(METRIC_NETWORK_OUTBOUND, $response->getSize())
-            ->addMetric(METRIC_EXECUTIONS, 1)
-            ->addMetric($metricTypeExecutions, 1)
-            ->addMetric($metricTypeIdExecutions, 1)
-            ->addMetric(METRIC_EXECUTIONS_COMPUTE, $compute) // per project
-            ->addMetric($metricTypeExecutionsCompute, $compute) // per function
-            ->addMetric($metricTypeIdExecutionsCompute, $compute) // per function
-            ->addMetric(METRIC_EXECUTIONS_MB_SECONDS, $mbSeconds)
-            ->addMetric($metricTypeExecutionsMbSeconds, $mbSeconds)
-            ->addMetric($metricTypeIdExecutionsMBSeconds, $mbSeconds)
-            ->setProject($project)
-            ->trigger();
+        $bus->dispatch(new RequestCompleted(
+            project: $project->getArrayCopy(),
+            request: $request,
+            response: $response,
+            deployment: $deployment->getArrayCopy(),
+        ));
 
         /* cleanup */
         if ($executionsRetentionCount > 0 && ENABLE_EXECUTIONS_LIMIT_ON_ROUTE) {
@@ -882,9 +827,8 @@ Http::init()
     ->inject('locale')
     ->inject('localeCodes')
     ->inject('geodb')
-    ->inject('queueForStatsUsage')
     ->inject('queueForEvents')
-    ->inject('queueForExecutions')
+    ->inject('bus')
     ->inject('executor')
     ->inject('platform')
     ->inject('isResourceBlocked')
@@ -895,7 +839,7 @@ Http::init()
     ->inject('authorization')
     ->inject('queueForDeletes')
     ->inject('executionsRetentionCount')
-    ->action(function (Http $utopia, SwooleRequest $swooleRequest, Request $request, Response $response, Log $log, Document $project, Database $dbForPlatform, callable $getProjectDB, Locale $locale, array $localeCodes, Reader $geodb, StatsUsage $queueForStatsUsage, Event $queueForEvents, Execution $queueForExecutions, Executor $executor, array $platform, callable $isResourceBlocked, string $previewHostname, Document $devKey, ?Key $apiKey, Cors $cors, Authorization $authorization, DeleteEvent $queueForDeletes, int $executionsRetentionCount) {
+    ->action(function (Http $utopia, SwooleRequest $swooleRequest, Request $request, Response $response, Log $log, Document $project, Database $dbForPlatform, callable $getProjectDB, Locale $locale, array $localeCodes, Reader $geodb, Event $queueForEvents, Bus $bus, Executor $executor, array $platform, callable $isResourceBlocked, string $previewHostname, Document $devKey, ?Key $apiKey, Cors $cors, Authorization $authorization, DeleteEvent $queueForDeletes, int $executionsRetentionCount) {
         /*
         * Appwrite Router
         */
@@ -903,7 +847,7 @@ Http::init()
         $platformHostnames = $platform['hostnames'] ?? [];
         // Only run Router when external domain
         if (!\in_array($hostname, $platformHostnames) || !empty($previewHostname)) {
-            if (router($utopia, $dbForPlatform, $getProjectDB, $swooleRequest, $request, $response, $log, $queueForEvents, $queueForStatsUsage, $queueForExecutions, $executor, $geodb, $isResourceBlocked, $platform, $previewHostname, $authorization, $apiKey, $queueForDeletes, $executionsRetentionCount)) {
+            if (router($utopia, $dbForPlatform, $getProjectDB, $swooleRequest, $request, $response, $log, $queueForEvents, $bus, $executor, $geodb, $isResourceBlocked, $platform, $previewHostname, $authorization, $apiKey, $queueForDeletes, $executionsRetentionCount)) {
                 $utopia->getRoute()?->label('router', true);
             }
         }
@@ -1182,8 +1126,7 @@ Http::options()
     ->inject('dbForPlatform')
     ->inject('getProjectDB')
     ->inject('queueForEvents')
-    ->inject('queueForStatsUsage')
-    ->inject('queueForExecutions')
+    ->inject('bus')
     ->inject('executor')
     ->inject('geodb')
     ->inject('isResourceBlocked')
@@ -1196,14 +1139,14 @@ Http::options()
     ->inject('authorization')
     ->inject('queueForDeletes')
     ->inject('executionsRetentionCount')
-    ->action(function (Http $utopia, SwooleRequest $swooleRequest, Request $request, Response $response, Log $log, Database $dbForPlatform, callable $getProjectDB, Event $queueForEvents, StatsUsage $queueForStatsUsage, Execution $queueForExecutions, Executor $executor, Reader $geodb, callable $isResourceBlocked, array $platform, string $previewHostname, Document $project, Document $devKey, ?Key $apiKey, Cors $cors, Authorization $authorization, DeleteEvent $queueForDeletes, int $executionsRetentionCount) {
+    ->action(function (Http $utopia, SwooleRequest $swooleRequest, Request $request, Response $response, Log $log, Database $dbForPlatform, callable $getProjectDB, Event $queueForEvents, Bus $bus, Executor $executor, Reader $geodb, callable $isResourceBlocked, array $platform, string $previewHostname, Document $project, Document $devKey, ?Key $apiKey, Cors $cors, Authorization $authorization, DeleteEvent $queueForDeletes, int $executionsRetentionCount) {
         /*
         * Appwrite Router
         */
         $platformHostnames = $platform['hostnames'] ?? [];
         // Only run Router when external domain
         if (!in_array($request->getHostname(), $platformHostnames) || !empty($previewHostname)) {
-            if (router($utopia, $dbForPlatform, $getProjectDB, $swooleRequest, $request, $response, $log, $queueForEvents, $queueForStatsUsage, $queueForExecutions, $executor, $geodb, $isResourceBlocked, $platform, $previewHostname, $authorization, $apiKey, $queueForDeletes, $executionsRetentionCount)) {
+            if (router($utopia, $dbForPlatform, $getProjectDB, $swooleRequest, $request, $response, $log, $queueForEvents, $bus, $executor, $geodb, $isResourceBlocked, $platform, $previewHostname, $authorization, $apiKey, $queueForDeletes, $executionsRetentionCount)) {
                 $utopia->getRoute()?->label('router', true);
             }
         }
@@ -1219,12 +1162,11 @@ Http::options()
         /** OPTIONS requests in utopia do not execute shutdown handlers, as a result we need to track the OPTIONS requests explicitly
          * @see https://github.com/utopia-php/http/blob/0.33.16/src/App.php#L825-L855
          */
-        $queueForStatsUsage
-            ->addMetric(METRIC_NETWORK_REQUESTS, 1)
-            ->addMetric(METRIC_NETWORK_INBOUND, $request->getSize())
-            ->addMetric(METRIC_NETWORK_OUTBOUND, $response->getSize())
-            ->setProject($project)
-            ->trigger();
+        $bus->dispatch(new RequestCompleted(
+            project: $project->getArrayCopy(),
+            request: $request,
+            response: $response,
+        ));
     });
 
 Http::error()
@@ -1235,10 +1177,10 @@ Http::error()
     ->inject('project')
     ->inject('logger')
     ->inject('log')
-    ->inject('queueForStatsUsage')
+    ->inject('bus')
     ->inject('devKey')
     ->inject('authorization')
-    ->action(function (Throwable $error, Http $utopia, Request $request, Response $response, Document $project, ?Logger $logger, Log $log, StatsUsage $queueForStatsUsage, Document $devKey, Authorization $authorization) {
+    ->action(function (Throwable $error, Http $utopia, Request $request, Response $response, Document $project, ?Logger $logger, Log $log, Bus $bus, Document $devKey, Authorization $authorization) {
         $version = System::getEnv('_APP_VERSION', 'UNKNOWN');
         $route = $utopia->getRoute();
         $class = \get_class($error);
@@ -1311,21 +1253,12 @@ Http::error()
          */
         if (!$publish && $project->getId() !== 'console') {
             if (!DBUser::isPrivileged($authorization->getRoles())) {
-                $fileSize = 0;
-                $file = $request->getFiles('file');
-                if (!empty($file)) {
-                    $fileSize = (\is_array($file['size']) && isset($file['size'][0])) ? $file['size'][0] : $file['size'];
-                }
-
-                $queueForStatsUsage
-                    ->addMetric(METRIC_NETWORK_REQUESTS, 1)
-                    ->addMetric(METRIC_NETWORK_INBOUND, $request->getSize() + $fileSize)
-                    ->addMetric(METRIC_NETWORK_OUTBOUND, $response->getSize());
+                $bus->dispatch(new RequestCompleted(
+                    project: $project->getArrayCopy(),
+                    request: $request,
+                    response: $response,
+                ));
             }
-
-            $queueForStatsUsage
-                ->setProject($project)
-                ->trigger();
         }
 
         if ($logger && $publish) {
@@ -1572,8 +1505,7 @@ Http::get('/robots.txt')
     ->inject('dbForPlatform')
     ->inject('getProjectDB')
     ->inject('queueForEvents')
-    ->inject('queueForStatsUsage')
-    ->inject('queueForExecutions')
+    ->inject('bus')
     ->inject('executor')
     ->inject('geodb')
     ->inject('isResourceBlocked')
@@ -1583,13 +1515,13 @@ Http::get('/robots.txt')
     ->inject('authorization')
     ->inject('queueForDeletes')
     ->inject('executionsRetentionCount')
-    ->action(function (Http $utopia, SwooleRequest $swooleRequest, Request $request, Response $response, Log $log, Database $dbForPlatform, callable $getProjectDB, Event $queueForEvents, StatsUsage $queueForStatsUsage, Execution $queueForExecutions, Executor $executor, Reader $geodb, callable $isResourceBlocked, array $platform, string $previewHostname, ?Key $apiKey, Authorization $authorization, DeleteEvent $queueForDeletes, int $executionsRetentionCount) {
+    ->action(function (Http $utopia, SwooleRequest $swooleRequest, Request $request, Response $response, Log $log, Database $dbForPlatform, callable $getProjectDB, Event $queueForEvents, Bus $bus, Executor $executor, Reader $geodb, callable $isResourceBlocked, array $platform, string $previewHostname, ?Key $apiKey, Authorization $authorization, DeleteEvent $queueForDeletes, int $executionsRetentionCount) {
         $platformHostnames = $platform['hostnames'] ?? [];
         if (in_array($request->getHostname(), $platformHostnames) || !empty($previewHostname)) {
             $template = new View(__DIR__ . '/../views/general/robots.phtml');
             $response->text($template->render(false));
         } else {
-            if (router($utopia, $dbForPlatform, $getProjectDB, $swooleRequest, $request, $response, $log, $queueForEvents, $queueForStatsUsage, $queueForExecutions, $executor, $geodb, $isResourceBlocked, $platform, $previewHostname, $authorization, $apiKey, $queueForDeletes, $executionsRetentionCount)) {
+            if (router($utopia, $dbForPlatform, $getProjectDB, $swooleRequest, $request, $response, $log, $queueForEvents, $bus, $executor, $geodb, $isResourceBlocked, $platform, $previewHostname, $authorization, $apiKey, $queueForDeletes, $executionsRetentionCount)) {
                 $utopia->getRoute()?->label('router', true);
             }
         }
@@ -1607,8 +1539,7 @@ Http::get('/humans.txt')
     ->inject('dbForPlatform')
     ->inject('getProjectDB')
     ->inject('queueForEvents')
-    ->inject('queueForStatsUsage')
-    ->inject('queueForExecutions')
+    ->inject('bus')
     ->inject('executor')
     ->inject('geodb')
     ->inject('isResourceBlocked')
@@ -1618,13 +1549,13 @@ Http::get('/humans.txt')
     ->inject('authorization')
     ->inject('queueForDeletes')
     ->inject('executionsRetentionCount')
-    ->action(function (Http $utopia, SwooleRequest $swooleRequest, Request $request, Response $response, Log $log, Database $dbForPlatform, callable $getProjectDB, Event $queueForEvents, StatsUsage $queueForStatsUsage, Execution $queueForExecutions, Executor $executor, Reader $geodb, callable $isResourceBlocked, array $platform, string $previewHostname, ?Key $apiKey, Authorization $authorization, DeleteEvent $queueForDeletes, int $executionsRetentionCount) {
+    ->action(function (Http $utopia, SwooleRequest $swooleRequest, Request $request, Response $response, Log $log, Database $dbForPlatform, callable $getProjectDB, Event $queueForEvents, Bus $bus, Executor $executor, Reader $geodb, callable $isResourceBlocked, array $platform, string $previewHostname, ?Key $apiKey, Authorization $authorization, DeleteEvent $queueForDeletes, int $executionsRetentionCount) {
         $platformHostnames = $platform['hostnames'] ?? [];
         if (in_array($request->getHostname(), $platformHostnames) || !empty($previewHostname)) {
             $template = new View(__DIR__ . '/../views/general/humans.phtml');
             $response->text($template->render(false));
         } else {
-            if (router($utopia, $dbForPlatform, $getProjectDB, $swooleRequest, $request, $response, $log, $queueForEvents, $queueForStatsUsage, $queueForExecutions, $executor, $geodb, $isResourceBlocked, $platform, $previewHostname, $authorization, $apiKey, $queueForDeletes, $executionsRetentionCount)) {
+            if (router($utopia, $dbForPlatform, $getProjectDB, $swooleRequest, $request, $response, $log, $queueForEvents, $bus, $executor, $geodb, $isResourceBlocked, $platform, $previewHostname, $authorization, $apiKey, $queueForDeletes, $executionsRetentionCount)) {
                 $utopia->getRoute()?->label('router', true);
             }
         }

app/controllers/shared/api.php

@@ -2,6 +2,7 @@
 
 use Appwrite\Auth\Key;
 use Appwrite\Auth\MFA\Type\TOTP;
+use Appwrite\Bus\Events\RequestCompleted;
 use Appwrite\Event\Audit;
 use Appwrite\Event\Build;
 use Appwrite\Event\Database as EventDatabase;
@@ -21,6 +22,7 @@ use Appwrite\Utopia\Database\Documents\User;
 use Appwrite\Utopia\Request;
 use Appwrite\Utopia\Response;
 use Utopia\Abuse\Abuse;
+use Utopia\Bus\Bus;
 use Utopia\Cache\Adapter\Filesystem;
 use Utopia\Cache\Cache;
 use Utopia\Config\Config;
@@ -746,7 +748,8 @@ Http::shutdown()
     ->inject('authorization')
     ->inject('timelimit')
     ->inject('eventProcessor')
-    ->action(function (Http $utopia, Request $request, Response $response, Document $project, User $user, Event $queueForEvents, Audit $queueForAudits, StatsUsage $queueForStatsUsage, Delete $queueForDeletes, EventDatabase $queueForDatabase, Build $queueForBuilds, Messaging $queueForMessaging, Func $queueForFunctions, Event $queueForWebhooks, Realtime $queueForRealtime, Database $dbForProject, Authorization $authorization, callable $timelimit, EventProcessor $eventProcessor) use ($parseLabel) {
+    ->inject('bus')
+    ->action(function (Http $utopia, Request $request, Response $response, Document $project, User $user, Event $queueForEvents, Audit $queueForAudits, StatsUsage $queueForStatsUsage, Delete $queueForDeletes, EventDatabase $queueForDatabase, Build $queueForBuilds, Messaging $queueForMessaging, Func $queueForFunctions, Event $queueForWebhooks, Realtime $queueForRealtime, Database $dbForProject, Authorization $authorization, callable $timelimit, EventProcessor $eventProcessor, Bus $bus) use ($parseLabel) {
 
         $responsePayload = $response->getPayload();
 
@@ -958,16 +961,11 @@ Http::shutdown()
 
         if ($project->getId() !== 'console') {
             if (!User::isPrivileged($authorization->getRoles())) {
-                $fileSize = 0;
-                $file = $request->getFiles('file');
-                if (!empty($file)) {
-                    $fileSize = (\is_array($file['size']) && isset($file['size'][0])) ? $file['size'][0] : $file['size'];
-                }
-
-                $queueForStatsUsage
-                    ->addMetric(METRIC_NETWORK_REQUESTS, 1)
-                    ->addMetric(METRIC_NETWORK_INBOUND, $request->getSize() + $fileSize)
-                    ->addMetric(METRIC_NETWORK_OUTBOUND, $response->getSize());
+                $bus->dispatch(new RequestCompleted(
+                    project: $project->getArrayCopy(),
+                    request: $request,
+                    response: $response,
+                ));
             }
 
             $queueForStatsUsage

app/http.php

@@ -188,6 +188,10 @@ $http->on(Constant::EVENT_AFTER_RELOAD, function ($server) {
     Console::success('Reload completed...');
 });
 
+Http::setResource('bus', function ($register, $utopia) {
+    return $register->get('bus')->setResolver(fn (string $name) => $utopia->getResource($name));
+}, ['register', 'utopia']);
+
 include __DIR__ . '/controllers/general.php';
 
 function createDatabase(Http $app, string $resourceKey, string $dbName, array $collections, mixed $pools, ?callable $extraSetup = null): void

app/init/registers.php

@@ -449,3 +449,11 @@ $register->set('promiseAdapter', function () {
 $register->set('hooks', function () {
     return new Hooks();
 });
+$listeners = require __DIR__ . '/../listeners.php';
+$register->set('bus', function () use ($listeners) {
+    $bus = new \Utopia\Bus\Bus();
+    foreach ($listeners as $listener) {
+        $bus->subscribe($listener);
+    }
+    return $bus;
+});

app/init/resources.php

@@ -10,7 +10,6 @@ use Appwrite\Event\Certificate;
 use Appwrite\Event\Database as EventDatabase;
 use Appwrite\Event\Delete;
 use Appwrite\Event\Event;
-use Appwrite\Event\Execution;
 use Appwrite\Event\Func;
 use Appwrite\Event\Mail;
 use Appwrite\Event\Messaging;
@@ -160,9 +159,6 @@ Http::setResource('queueForAudits', function (Publisher $publisher) {
 Http::setResource('queueForFunctions', function (Publisher $publisher) {
     return new Func($publisher);
 }, ['publisher']);
-Http::setResource('queueForExecutions', function (Publisher $publisher) {
-    return new Execution($publisher);
-}, ['publisher']);
 Http::setResource('eventProcessor', function () {
     return new EventProcessor();
 }, []);

app/listeners.php

@@ -0,0 +1,9 @@
+<?php
+
+use Appwrite\Bus\Listeners\Log;
+use Appwrite\Bus\Listeners\Usage;
+
+return [
+    new Log(),
+    new Usage(),
+];

app/worker.php

@@ -9,7 +9,6 @@ use Appwrite\Event\Certificate;
 use Appwrite\Event\Database as EventDatabase;
 use Appwrite\Event\Delete;
 use Appwrite\Event\Event;
-use Appwrite\Event\Execution;
 use Appwrite\Event\Func;
 use Appwrite\Event\Mail;
 use Appwrite\Event\Messaging;
@@ -355,9 +354,6 @@ Server::setResource('queueForFunctions', function (Publisher $publisher) {
     return new Func($publisher);
 }, ['publisher']);
 
-Server::setResource('queueForExecutions', function (Publisher $publisher) {
-    return new Execution($publisher);
-}, ['publisher']);
 
 Server::setResource('queueForRealtime', function () {
     return new Realtime();
@@ -542,6 +538,10 @@ try {
 
 $worker = $platform->getWorker();
 
+Server::setResource('bus', function ($register) use ($worker) {
+    return $register->get('bus')->setResolver(fn (string $name) => $worker->getResource($name));
+}, ['register']);
+
 $worker
     ->error()
     ->inject('error')

composer.json

@@ -19,7 +19,8 @@
     "autoload": {
         "psr-4": {
             "Appwrite\\": "src/Appwrite",
-            "Executor\\": "src/Executor"
+            "Executor\\": "src/Executor",
+            "Utopia\\Bus\\": "src/Utopia/Bus"
         }
     },
     "autoload-dev": {

phpstan.neon

@@ -1,7 +1,11 @@
 parameters:
   level: 8
   paths:
+    - src/Utopia/Bus
+    - src/Appwrite/Bus
     - src/Appwrite/Transformation
+  bootstrapFiles:
+    - app/init/constants.php
   scanDirectories:
     - vendor/swoole/ide-helper
   excludePaths:

src/Appwrite/Bus/Events/ExecutionCompleted.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace Appwrite\Bus\Events;
+
+use Utopia\Bus\Event;
+
+class ExecutionCompleted implements Event
+{
+    /**
+     * @param array<string, mixed> $execution
+     * @param array<string, mixed> $project
+     * @param array<string, mixed> $spec
+     * @param array<string, mixed> $resource
+     */
+    public function __construct(
+        public readonly array $execution,
+        public readonly array $project,
+        public readonly array $spec = [],
+        public readonly array $resource = [],
+    ) {
+    }
+}

src/Appwrite/Bus/Events/RequestCompleted.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace Appwrite\Bus\Events;
+
+use Appwrite\Utopia\Request;
+use Appwrite\Utopia\Response;
+use Utopia\Bus\Event;
+
+class RequestCompleted implements Event
+{
+    /**
+     * @param array<string, mixed> $project
+     * @param array<string, mixed> $deployment
+     */
+    public function __construct(
+        public readonly array $project,
+        public readonly Request $request,
+        public readonly Response $response,
+        public readonly array $deployment = [],
+    ) {
+    }
+}

src/Appwrite/Bus/Listeners/Log.php

@@ -0,0 +1,39 @@
+<?php
+
+namespace Appwrite\Bus\Listeners;
+
+use Appwrite\Bus\Events\ExecutionCompleted;
+use Appwrite\Event\Execution;
+use Utopia\Bus\Listener;
+use Utopia\Database\Document;
+use Utopia\Queue\Publisher;
+
+class Log extends Listener
+{
+    public static function getName(): string
+    {
+        return 'log';
+    }
+
+    public static function getEvents(): array
+    {
+        return [ExecutionCompleted::class];
+    }
+
+    public function __construct()
+    {
+        $this
+            ->desc('Persists execution logs to database via queue')
+            ->inject('publisher')
+            ->callback($this->handle(...));
+    }
+
+    public function handle(ExecutionCompleted $event, Publisher $publisher): void
+    {
+        $queueForExecutions = new Execution($publisher);
+        $queueForExecutions
+            ->setExecution(new Document($event->execution))
+            ->setProject(new Document($event->project))
+            ->trigger();
+    }
+}

src/Appwrite/Bus/Listeners/Usage.php

@@ -0,0 +1,114 @@
+<?php
+
+namespace Appwrite\Bus\Listeners;
+
+use Appwrite\Bus\Events\ExecutionCompleted;
+use Appwrite\Bus\Events\RequestCompleted;
+use Appwrite\Event\StatsUsage;
+use Utopia\Bus\Event;
+use Utopia\Bus\Listener;
+use Utopia\Database\Document;
+use Utopia\Queue\Publisher;
+
+class Usage extends Listener
+{
+    public static function getName(): string
+    {
+        return 'usage';
+    }
+
+    public static function getEvents(): array
+    {
+        return [
+            ExecutionCompleted::class,
+            RequestCompleted::class,
+        ];
+    }
+
+    public function __construct()
+    {
+        $this
+            ->desc('Records usage metrics')
+            ->inject('publisherStatsUsage')
+            ->callback($this->handle(...));
+    }
+
+    public function handle(Event $event, Publisher $publisher): void
+    {
+        match (true) {
+            $event instanceof ExecutionCompleted => $this->handleExecutionCompleted($event, $publisher),
+            $event instanceof RequestCompleted => $this->handleRequestCompleted($event, $publisher),
+            default => null,
+        };
+    }
+
+    private function handleExecutionCompleted(ExecutionCompleted $event, Publisher $publisher): void
+    {
+        $execution = new Document($event->execution);
+        $resource = new Document($event->resource);
+
+        // Non-SSR sites don't record execution metrics
+        if ($execution->getAttribute('resourceType') === 'sites' && $resource->getAttribute('adapter') !== 'ssr') {
+            return;
+        }
+        $project = new Document($event->project);
+        $spec = $event->spec;
+
+        $resourceType = $execution->getAttribute('resourceType', '');
+        $resourceInternalId = $execution->getAttribute('resourceInternalId', '');
+        $duration = $execution->getAttribute('duration', 0);
+
+        $compute = (int)($duration * 1000);
+        $mbSeconds = (int)(($spec['memory'] ?? APP_COMPUTE_MEMORY_DEFAULT) * $duration * ($spec['cpus'] ?? APP_COMPUTE_CPUS_DEFAULT));
+
+        $queueForStatsUsage = new StatsUsage($publisher);
+        $queueForStatsUsage
+            ->setProject($project)
+            ->addMetric(METRIC_EXECUTIONS, 1)
+            ->addMetric(str_replace(['{resourceType}'], [$resourceType], METRIC_RESOURCE_TYPE_EXECUTIONS), 1)
+            ->addMetric(str_replace(['{resourceType}', '{resourceInternalId}'], [$resourceType, $resourceInternalId], METRIC_RESOURCE_TYPE_ID_EXECUTIONS), 1)
+            ->addMetric(METRIC_EXECUTIONS_COMPUTE, $compute)
+            ->addMetric(str_replace(['{resourceType}'], [$resourceType], METRIC_RESOURCE_TYPE_EXECUTIONS_COMPUTE), $compute)
+            ->addMetric(str_replace(['{resourceType}', '{resourceInternalId}'], [$resourceType, $resourceInternalId], METRIC_RESOURCE_TYPE_ID_EXECUTIONS_COMPUTE), $compute)
+            ->addMetric(METRIC_EXECUTIONS_MB_SECONDS, $mbSeconds)
+            ->addMetric(str_replace(['{resourceType}'], [$resourceType], METRIC_RESOURCE_TYPE_EXECUTIONS_MB_SECONDS), $mbSeconds)
+            ->addMetric(str_replace(['{resourceType}', '{resourceInternalId}'], [$resourceType, $resourceInternalId], METRIC_RESOURCE_TYPE_ID_EXECUTIONS_MB_SECONDS), $mbSeconds)
+            ->trigger();
+    }
+
+    private function handleRequestCompleted(RequestCompleted $event, Publisher $publisher): void
+    {
+        $fileSize = 0;
+        $file = $event->request->getFiles('file');
+        if (!empty($file)) {
+            $fileSize = (\is_array($file['size']) && isset($file['size'][0])) ? $file['size'][0] : $file['size'];
+        }
+
+        $project = new Document($event->project);
+        $deployment = new Document($event->deployment);
+        $queueForStatsUsage = new StatsUsage($publisher);
+
+        $inbound = $event->request->getSize() + $fileSize;
+        $outbound = $event->response->getSize();
+
+        $queueForStatsUsage->setProject($project);
+
+        if ($deployment->getAttribute('resourceType') === 'sites') {
+            $siteInternalId = $deployment->getAttribute('resourceInternalId', '');
+            $queueForStatsUsage
+                ->addMetric(METRIC_SITES_REQUESTS, 1)
+                ->addMetric(METRIC_SITES_INBOUND, $inbound)
+                ->addMetric(METRIC_SITES_OUTBOUND, $outbound)
+                ->addMetric(str_replace('{siteInternalId}', $siteInternalId, METRIC_SITES_ID_REQUESTS), 1)
+                ->addMetric(str_replace('{siteInternalId}', $siteInternalId, METRIC_SITES_ID_INBOUND), $inbound)
+                ->addMetric(str_replace('{siteInternalId}', $siteInternalId, METRIC_SITES_ID_OUTBOUND), $outbound);
+        } else {
+            $queueForStatsUsage
+                ->addMetric(METRIC_NETWORK_REQUESTS, 1)
+                ->addMetric(METRIC_NETWORK_INBOUND, $inbound)
+                ->addMetric(METRIC_NETWORK_OUTBOUND, $outbound);
+        }
+
+        $queueForStatsUsage->trigger();
+    }
+}

src/Appwrite/Platform/Workers/Functions.php

@@ -3,15 +3,15 @@
 namespace Appwrite\Platform\Workers;
 
 use Ahc\Jwt\JWT;
+use Appwrite\Bus\Events\ExecutionCompleted;
 use Appwrite\Event\Event;
-use Appwrite\Event\Execution as ExecutionEvent;
 use Appwrite\Event\Func;
 use Appwrite\Event\Realtime;
-use Appwrite\Event\StatsUsage;
 use Appwrite\Event\Webhook;
 use Appwrite\Extend\Exception as AppwriteException;
 use Appwrite\Utopia\Response\Model\Execution;
 use Executor\Executor;
+use Utopia\Bus\Bus;
 use Utopia\Config\Config;
 use Utopia\Console;
 use Utopia\Database\Database;
@@ -47,8 +47,7 @@ class Functions extends Action
             ->inject('queueForFunctions')
             ->inject('queueForRealtime')
             ->inject('queueForEvents')
-            ->inject('queueForStatsUsage')
-            ->inject('queueForExecutions')
+            ->inject('bus')
             ->inject('log')
             ->inject('executor')
             ->inject('isResourceBlocked')
@@ -63,8 +62,7 @@ class Functions extends Action
         Func $queueForFunctions,
         Realtime $queueForRealtime,
         Event $queueForEvents,
-        StatsUsage $queueForStatsUsage,
-        ExecutionEvent $queueForExecutions,
+        Bus $bus,
         Log $log,
         Executor $executor,
         callable $isResourceBlocked
@@ -159,9 +157,8 @@ class Functions extends Action
                         queueForWebhooks: $queueForWebhooks,
                         queueForFunctions: $queueForFunctions,
                         queueForRealtime: $queueForRealtime,
-                        queueForStatsUsage: $queueForStatsUsage,
                         queueForEvents: $queueForEvents,
-                        queueForExecutions: $queueForExecutions,
+                        bus: $bus,
                         project: $project,
                         function: $function,
                         executor:  $executor,
@@ -204,9 +201,8 @@ class Functions extends Action
                     queueForWebhooks: $queueForWebhooks,
                     queueForFunctions: $queueForFunctions,
                     queueForRealtime: $queueForRealtime,
-                    queueForStatsUsage: $queueForStatsUsage,
                     queueForEvents: $queueForEvents,
-                    queueForExecutions: $queueForExecutions,
+                    bus: $bus,
                     project: $project,
                     function: $function,
                     executor:  $executor,
@@ -231,9 +227,8 @@ class Functions extends Action
                     queueForWebhooks: $queueForWebhooks,
                     queueForFunctions: $queueForFunctions,
                     queueForRealtime: $queueForRealtime,
-                    queueForStatsUsage: $queueForStatsUsage,
                     queueForEvents: $queueForEvents,
-                    queueForExecutions: $queueForExecutions,
+                    bus: $bus,
                     project: $project,
                     function: $function,
                     executor:  $executor,
@@ -267,7 +262,7 @@ class Functions extends Action
     private function fail(
         string $message,
         Document $project,
-        ExecutionEvent $queueForExecutions,
+        Bus $bus,
         Document $function,
         string $trigger,
         string $path,
@@ -310,10 +305,10 @@ class Functions extends Action
             'duration' => 0.0,
         ]);
 
-        $queueForExecutions
-            ->setExecution($execution)
-            ->setProject($project)
-            ->trigger();
+        $bus->dispatch(new ExecutionCompleted(
+            execution: $execution->getArrayCopy(),
+            project: $project->getArrayCopy(),
+        ));
     }
 
     /**
@@ -321,7 +316,6 @@ class Functions extends Action
      * @param Database $dbForProject
      * @param Func $queueForFunctions
      * @param Realtime $queueForRealtime
-     * @param StatsUsage $queueForStatsUsage
      * @param Event $queueForEvents
      * @param Document $project
      * @param Document $function
@@ -344,9 +338,8 @@ class Functions extends Action
         Webhook $queueForWebhooks,
         Func $queueForFunctions,
         Realtime $queueForRealtime,
-        StatsUsage $queueForStatsUsage,
         Event $queueForEvents,
-        ExecutionEvent $queueForExecutions,
+        Bus $bus,
         Document $project,
         Document $function,
         Executor $executor,
@@ -374,19 +367,19 @@ class Functions extends Action
 
         if ($deployment->getAttribute('resourceId') !== $functionId) {
             $errorMessage = 'The execution could not be completed because a corresponding deployment was not found. A function deployment needs to be created before it can be executed. Please create a deployment for your function and try again.';
-            $this->fail($errorMessage, $project, $queueForExecutions, $function, $trigger, $path, $method, $user, $jwt, $event);
+            $this->fail($errorMessage, $project, $bus, $function, $trigger, $path, $method, $user, $jwt, $event);
             return;
         }
 
         if ($deployment->isEmpty()) {
             $errorMessage = 'The execution could not be completed because a corresponding deployment was not found. A function deployment needs to be created before it can be executed. Please create a deployment for your function and try again.';
-            $this->fail($errorMessage, $project, $queueForExecutions, $function, $trigger, $path, $method, $user, $jwt, $event);
+            $this->fail($errorMessage, $project, $bus, $function, $trigger, $path, $method, $user, $jwt, $event);
             return;
         }
 
         if ($deployment->getAttribute('status') !== 'ready') {
             $errorMessage = 'The execution could not be completed because the build is not ready. Please wait for the build to complete and try again.';
-            $this->fail($errorMessage, $project, $queueForExecutions, $function, $trigger, $path, $method, $user, $jwt, $event);
+            $this->fail($errorMessage, $project, $bus, $function, $trigger, $path, $method, $user, $jwt, $event);
             return;
         }
 
@@ -593,26 +586,12 @@ class Functions extends Action
             $error = $th->getMessage();
             $errorCode = $th->getCode();
         } finally {
-            /** Persist final execution status */
-            $queueForExecutions
-                ->setExecution($execution)
-                ->setProject($project)
-                ->trigger();
-
-            /** Trigger usage queue */
-            $queueForStatsUsage
-                ->setProject($project)
-                ->addMetric(METRIC_EXECUTIONS, 1)
-                ->addMetric(str_replace(['{resourceType}'], [RESOURCE_TYPE_FUNCTIONS], METRIC_RESOURCE_TYPE_EXECUTIONS), 1)
-                ->addMetric(str_replace(['{resourceType}', '{resourceInternalId}'], [RESOURCE_TYPE_FUNCTIONS, $function->getSequence()], METRIC_RESOURCE_TYPE_ID_EXECUTIONS), 1)
-                ->addMetric(METRIC_EXECUTIONS_COMPUTE, (int)($execution->getAttribute('duration') * 1000))// per project
-                ->addMetric(str_replace(['{resourceType}'], [RESOURCE_TYPE_FUNCTIONS], METRIC_RESOURCE_TYPE_EXECUTIONS_COMPUTE), (int)($execution->getAttribute('duration') * 1000))
-                ->addMetric(str_replace(['{resourceType}', '{resourceInternalId}'], [RESOURCE_TYPE_FUNCTIONS, $function->getSequence()], METRIC_RESOURCE_TYPE_ID_EXECUTIONS_COMPUTE), (int)($execution->getAttribute('duration') * 1000))
-                ->addMetric(METRIC_EXECUTIONS_MB_SECONDS, (int)(($spec['memory'] ?? APP_COMPUTE_MEMORY_DEFAULT) * $execution->getAttribute('duration', 0) * ($spec['cpus'] ?? APP_COMPUTE_CPUS_DEFAULT)))
-                ->addMetric(str_replace(['{resourceType}'], [RESOURCE_TYPE_FUNCTIONS], METRIC_RESOURCE_TYPE_EXECUTIONS_MB_SECONDS), (int)(($spec['memory'] ?? APP_COMPUTE_MEMORY_DEFAULT) * $execution->getAttribute('duration', 0) * ($spec['cpus'] ?? APP_COMPUTE_CPUS_DEFAULT)))
-                ->addMetric(str_replace(['{resourceType}', '{resourceInternalId}'], [RESOURCE_TYPE_FUNCTIONS, $function->getSequence()], METRIC_RESOURCE_TYPE_ID_EXECUTIONS_MB_SECONDS), (int)(($spec['memory'] ?? APP_COMPUTE_MEMORY_DEFAULT) * $execution->getAttribute('duration', 0) * ($spec['cpus'] ?? APP_COMPUTE_CPUS_DEFAULT)))
-                ->trigger()
-            ;
+            /** Persist final execution status and record usage */
+            $bus->dispatch(new ExecutionCompleted(
+                execution: $execution->getArrayCopy(),
+                project: $project->getArrayCopy(),
+                spec: $spec,
+            ));
         }
 
         $executionModel = new Execution();

src/Utopia/Bus/Bus.php

@@ -0,0 +1,51 @@
+<?php
+
+namespace Utopia\Bus;
+
+use Utopia\Span\Span;
+
+class Bus
+{
+    /** @var array<class-string<Event>, Listener[]> */
+    private array $listeners = [];
+
+    /** @var ?\Closure(string): mixed */
+    private ?\Closure $resolver = null;
+
+    public function setResolver(callable $resolver): self
+    {
+        $this->resolver = $resolver(...);
+        return $this;
+    }
+
+    public function subscribe(Listener $listener): self
+    {
+        foreach ($listener::getEvents() as $event) {
+            $this->listeners[$event][] = $listener;
+        }
+        return $this;
+    }
+
+    public function dispatch(Event $event): void
+    {
+        if ($this->resolver === null) {
+            throw new \LogicException('Bus resolver must be set via setResolver() before dispatching events');
+        }
+
+        $resolver = $this->resolver;
+        $listeners = $this->listeners[$event::class] ?? [];
+
+        foreach ($listeners as $listener) {
+            $deps = array_map($resolver, $listener->getInjections());
+            Span::init('listener.' . $listener::getName());
+            Span::add('bus.event', $event::class);
+            try {
+                ($listener->getCallback())($event, ...$deps);
+            } catch (\Throwable $e) {
+                Span::error($e);
+            } finally {
+                Span::current()?->finish();
+            }
+        }
+    }
+}

src/Utopia/Bus/Event.php

@@ -0,0 +1,7 @@
+<?php
+
+namespace Utopia\Bus;
+
+interface Event
+{
+}

src/Utopia/Bus/Listener.php

@@ -0,0 +1,51 @@
+<?php
+
+namespace Utopia\Bus;
+
+abstract class Listener
+{
+    protected ?string $desc = null;
+    /** @var array<string> */
+    protected array $injections = [];
+    protected ?\Closure $callback = null;
+
+    abstract public static function getName(): string;
+
+    /**
+     * @return array<class-string<Event>>
+     */
+    abstract public static function getEvents(): array;
+
+    protected function desc(string $desc): self
+    {
+        $this->desc = $desc;
+        return $this;
+    }
+
+    protected function inject(string $injection): self
+    {
+        $this->injections[] = $injection;
+        return $this;
+    }
+
+    protected function callback(callable $callback): self
+    {
+        $this->callback = $callback(...);
+        return $this;
+    }
+
+    /** @return array<string> */
+    public function getInjections(): array
+    {
+        return $this->injections;
+    }
+
+    public function getCallback(): callable
+    {
+        if ($this->callback === null) {
+            throw new \LogicException(static::class . ' must set a callback via $this->callback()');
+        }
+
+        return $this->callback;
+    }
+}

tests/e2e/Services/Realtime/RealtimeBase.php

@@ -11,7 +11,8 @@ trait RealtimeBase
         array $channels = [],
         array $headers = [],
         ?string $projectId = null,
-        ?array $queries = null
+        ?array $queries = null,
+        int $timeout = 2
     ): WebSocketClient {
         if (is_null($projectId)) {
             $projectId = $this->getProject()['$id'];
@@ -63,7 +64,7 @@ trait RealtimeBase
             "ws://appwrite.test/v1/realtime?" . $queryString,
             [
                 "headers" => $headers,
-                "timeout" => 45,
+                "timeout" => $timeout,
             ]
         );
     }
@@ -74,9 +75,10 @@ trait RealtimeBase
      *
      * @param array $queryParams Custom query parameters (e.g., ['channels' => ['project'], 'project' => [...]])
      * @param array $headers HTTP headers
+     * @param int $timeout Timeout in seconds (default: 2)
      * @return WebSocketClient
      */
-    private function getWebsocketWithCustomQuery(array $queryParams, array $headers = []): WebSocketClient
+    private function getWebsocketWithCustomQuery(array $queryParams, array $headers = [], int $timeout = 2): WebSocketClient
     {
         $queryString = http_build_query($queryParams);
 
@@ -84,7 +86,7 @@ trait RealtimeBase
             "ws://appwrite.test/v1/realtime?" . $queryString,
             [
                 "headers" => $headers,
-                "timeout" => 45,
+                "timeout" => $timeout,
             ]
         );
     }

tests/e2e/Services/Realtime/RealtimeCustomClientQueryTest.php

@@ -2535,29 +2535,35 @@ class RealtimeCustomClientQueryTest extends Scope
         $projectId = 'console';
 
         // Subscribe without queries - should receive all events
-        $clientNoQuery = $this->getWebsocket(['tests'], [
-            'origin' => 'http://localhost',
-        ], $projectId);
+        $clientNoQuery = $this->getWebsocket(
+            channels: ['tests'],
+            headers: ['origin' => 'http://localhost'],
+            projectId: $projectId,
+            timeout: 5
+        );
 
         $response = json_decode($clientNoQuery->receive(), true);
         $this->assertEquals('connected', $response['type']);
 
         // Subscribe with matching query - should receive events
-        $clientWithMatchingQuery = $this->getWebsocket(['tests'], [
-            'origin' => 'http://localhost',
-        ], $projectId, [
-            Query::equal('response', ['WS:/v1/realtime:passed'])->toString(),
-        ]);
+        $clientWithMatchingQuery = $this->getWebsocket(
+            channels: ['tests'],
+            headers: ['origin' => 'http://localhost'],
+            projectId: $projectId,
+            queries: [Query::equal('response', ['WS:/v1/realtime:passed'])->toString()],
+            timeout: 5
+        );
 
         $response = json_decode($clientWithMatchingQuery->receive(), true);
         $this->assertEquals('connected', $response['type']);
 
         // Subscribe with non-matching query - should NOT receive events
-        $clientWithNonMatchingQuery = $this->getWebsocket(['tests'], [
-            'origin' => 'http://localhost',
-        ], $projectId, [
-            Query::equal('response', ['failed'])->toString(),
-        ]);
+        $clientWithNonMatchingQuery = $this->getWebsocket(
+            channels: ['tests'],
+            headers: ['origin' => 'http://localhost'],
+            projectId: $projectId,
+            queries: [Query::equal('response', ['failed'])->toString()]
+        );
 
         $response = json_decode($clientWithNonMatchingQuery->receive(), true);
         $this->assertEquals('connected', $response['type']);

tests/e2e/Services/Realtime/RealtimeCustomClientTest.php

@@ -2225,10 +2225,14 @@ class RealtimeCustomClientTest extends Scope
         $session = $user['session'] ?? '';
         $projectId = $this->getProject()['$id'];
 
-        $client = $this->getWebsocket(['executions'], [
-            'origin' => 'http://localhost',
-            'cookie' => 'a_session_' . $projectId . '=' . $session
-        ]);
+        $client = $this->getWebsocket(
+            channels: ['executions'],
+            headers: [
+                'origin' => 'http://localhost',
+                'cookie' => 'a_session_' . $projectId . '=' . $session
+            ],
+            timeout: 10
+        );
 
         $response = json_decode($client->receive(), true);