Refactor migration error handling to provide clearer connection error messages. Updated exception messages for migration sources to guide users on potential credential and network issues.

chore: bump utopia-php/migration to 1.6.3

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/api/migrations.php

@@ -717,22 +717,14 @@ Http::get('/v1/migrations/appwrite/report')
     ->inject('project')
     ->inject('user')
     ->action(function (array $resources, string $endpoint, string $projectID, string $key, Response $response) {
-
-        $appwrite = new Appwrite($projectID, $endpoint, $key);
-
         try {
+            $appwrite = new Appwrite($projectID, $endpoint, $key);
             $report = $appwrite->report($resources);
         } catch (\Throwable $e) {
-            switch ($e->getCode()) {
-                case 401:
-                    throw new Exception(Exception::GENERAL_UNAUTHORIZED_SCOPE, 'Source Error: ' . $e->getMessage());
-                case 429:
-                    throw new Exception(Exception::GENERAL_RATE_LIMIT_EXCEEDED, 'Source Error: Rate Limit Exceeded, Is your Cloud Provider blocking Appwrite\'s IP?');
-                case 500:
-                    throw new Exception(Exception::MIGRATION_PROVIDER_ERROR, 'Source Error: ' . $e->getMessage());
-            }
-
-            throw new Exception(Exception::MIGRATION_PROVIDER_ERROR, 'Source Error: ' . $e->getMessage());
+            throw new Exception(
+                Exception::MIGRATION_PROVIDER_ERROR,
+                'Unable to connect to the migration source. Please verify your credentials and ensure the source is reachable from this server. Check for network restrictions such as firewalls, IP allowlists, or outbound connectivity limits.'
+            );
         }
 
         $response
@@ -771,21 +763,14 @@ Http::get('/v1/migrations/firebase/report')
             throw new Exception(Exception::MIGRATION_PROVIDER_ERROR, 'Invalid Service Account JSON');
         }
 
-        $firebase = new Firebase($serviceAccount);
-
         try {
+            $firebase = new Firebase($serviceAccount);
             $report = $firebase->report($resources);
         } catch (\Throwable $e) {
-            switch ($e->getCode()) {
-                case 401:
-                    throw new Exception(Exception::GENERAL_UNAUTHORIZED_SCOPE, 'Source Error: ' . $e->getMessage());
-                case 429:
-                    throw new Exception(Exception::GENERAL_RATE_LIMIT_EXCEEDED, 'Source Error: Rate Limit Exceeded, Is your Cloud Provider blocking Appwrite\'s IP?');
-                case 500:
-                    throw new Exception(Exception::MIGRATION_PROVIDER_ERROR, 'Source Error: ' . $e->getMessage());
-            }
-
-            throw new Exception(Exception::MIGRATION_PROVIDER_ERROR, 'Source Error: ' . $e->getMessage());
+            throw new Exception(
+                Exception::MIGRATION_PROVIDER_ERROR,
+                'Unable to connect to the migration source. Please verify your credentials and ensure the source is reachable from this server. Check for network restrictions such as firewalls, IP allowlists, or outbound connectivity limits.'
+            );
         }
 
         $response
@@ -820,21 +805,14 @@ Http::get('/v1/migrations/supabase/report')
     ->inject('response')
     ->inject('dbForProject')
     ->action(function (array $resources, string $endpoint, string $apiKey, string $databaseHost, string $username, string $password, int $port, Response $response) {
-        $supabase = new Supabase($endpoint, $apiKey, $databaseHost, 'postgres', $username, $password, $port);
-
         try {
+            $supabase = new Supabase($endpoint, $apiKey, $databaseHost, 'postgres', $username, $password, $port);
             $report = $supabase->report($resources);
         } catch (\Throwable $e) {
-            switch ($e->getCode()) {
-                case 401:
-                    throw new Exception(Exception::GENERAL_UNAUTHORIZED_SCOPE, 'Source Error: ' . $e->getMessage());
-                case 429:
-                    throw new Exception(Exception::GENERAL_RATE_LIMIT_EXCEEDED, 'Source Error: Rate Limit Exceeded, Is your Cloud Provider blocking Appwrite\'s IP?');
-                case 500:
-                    throw new Exception(Exception::MIGRATION_PROVIDER_ERROR, 'Source Error: ' . $e->getMessage());
-            }
-
-            throw new Exception(Exception::MIGRATION_PROVIDER_ERROR, 'Source Error: ' . $e->getMessage());
+            throw new Exception(
+                Exception::MIGRATION_PROVIDER_ERROR,
+                'Unable to connect to the migration source. Please verify your credentials and ensure the source is reachable from this server. Check for network restrictions such as firewalls, IP allowlists, or outbound connectivity limits.'
+            );
         }
 
         $response
@@ -869,21 +847,14 @@ Http::get('/v1/migrations/nhost/report')
     ->param('port', 5432, new Integer(true), 'Source\'s Database Port.', true)
     ->inject('response')
     ->action(function (array $resources, string $subdomain, string $region, string $adminSecret, string $database, string $username, string $password, int $port, Response $response) {
-        $nhost = new NHost($subdomain, $region, $adminSecret, $database, $username, $password, $port);
-
         try {
+            $nhost = new NHost($subdomain, $region, $adminSecret, $database, $username, $password, $port);
             $report = $nhost->report($resources);
         } catch (\Throwable $e) {
-            switch ($e->getCode()) {
-                case 401:
-                    throw new Exception(Exception::GENERAL_UNAUTHORIZED_SCOPE, 'Source Error: ' . $e->getMessage());
-                case 429:
-                    throw new Exception(Exception::GENERAL_RATE_LIMIT_EXCEEDED, 'Source Error: Rate Limit Exceeded, Is your Cloud Provider blocking Appwrite\'s IP?');
-                case 500:
-                    throw new Exception(Exception::MIGRATION_PROVIDER_ERROR, 'Source Error: ' . $e->getMessage());
-            }
-
-            throw new Exception(Exception::MIGRATION_PROVIDER_ERROR, 'Source Error: ' . $e->getMessage());
+            throw new Exception(
+                Exception::MIGRATION_PROVIDER_ERROR,
+                'Unable to connect to the migration source. Please verify your credentials and ensure the source is reachable from this server. Check for network restrictions such as firewalls, IP allowlists, or outbound connectivity limits.'
+            );
         }
 
         $response

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;
@@ -35,6 +35,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;
@@ -62,7 +63,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)) {
@@ -706,10 +707,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(),
+                ));
             }
         }
 
@@ -754,70 +757,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) {
@@ -881,9 +826,8 @@ Http::init()
     ->inject('locale')
     ->inject('localeCodes')
     ->inject('geodb')
-    ->inject('queueForStatsUsage')
     ->inject('queueForEvents')
-    ->inject('queueForExecutions')
+    ->inject('bus')
     ->inject('executor')
     ->inject('platform')
     ->inject('isResourceBlocked')
@@ -894,7 +838,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
         */
@@ -902,7 +846,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);
             }
         }
@@ -1178,8 +1122,7 @@ Http::options()
     ->inject('dbForPlatform')
     ->inject('getProjectDB')
     ->inject('queueForEvents')
-    ->inject('queueForStatsUsage')
-    ->inject('queueForExecutions')
+    ->inject('bus')
     ->inject('executor')
     ->inject('geodb')
     ->inject('isResourceBlocked')
@@ -1192,14 +1135,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);
             }
         }
@@ -1215,12 +1158,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()
@@ -1231,10 +1173,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);
@@ -1307,21 +1249,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) {
@@ -1568,8 +1501,7 @@ Http::get('/robots.txt')
     ->inject('dbForPlatform')
     ->inject('getProjectDB')
     ->inject('queueForEvents')
-    ->inject('queueForStatsUsage')
-    ->inject('queueForExecutions')
+    ->inject('bus')
     ->inject('executor')
     ->inject('geodb')
     ->inject('isResourceBlocked')
@@ -1579,13 +1511,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);
             }
         }
@@ -1603,8 +1535,7 @@ Http::get('/humans.txt')
     ->inject('dbForPlatform')
     ->inject('getProjectDB')
     ->inject('queueForEvents')
-    ->inject('queueForStatsUsage')
-    ->inject('queueForExecutions')
+    ->inject('bus')
     ->inject('executor')
     ->inject('geodb')
     ->inject('isResourceBlocked')
@@ -1614,13 +1545,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": {

composer.lock

@@ -4517,16 +4517,16 @@
         },
         {
             "name": "utopia-php/migration",
-            "version": "1.6.2",
+            "version": "1.6.3",
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/migration.git",
-                "reference": "037bf4b3813d44f1b0990bc124e35b501ed27fca"
+                "reference": "c2d016944cb029fa5ff822ceee704785a06ef289"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/migration/zipball/037bf4b3813d44f1b0990bc124e35b501ed27fca",
-                "reference": "037bf4b3813d44f1b0990bc124e35b501ed27fca",
+                "url": "https://api.github.com/repos/utopia-php/migration/zipball/c2d016944cb029fa5ff822ceee704785a06ef289",
+                "reference": "c2d016944cb029fa5ff822ceee704785a06ef289",
                 "shasum": ""
             },
             "require": {
@@ -4566,9 +4566,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/migration/issues",
-                "source": "https://github.com/utopia-php/migration/tree/1.6.2"
+                "source": "https://github.com/utopia-php/migration/tree/1.6.3"
             },
-            "time": "2026-02-25T12:00:11+00:00"
+            "time": "2026-03-04T07:08:22+00:00"
         },
         {
             "name": "utopia-php/mongo",

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
@@ -158,9 +156,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,
@@ -203,9 +200,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,
@@ -230,9 +226,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,
@@ -266,7 +261,7 @@ class Functions extends Action
     private function fail(
         string $message,
         Document $project,
-        ExecutionEvent $queueForExecutions,
+        Bus $bus,
         Document $function,
         string $trigger,
         string $path,
@@ -309,10 +304,10 @@ class Functions extends Action
             'duration' => 0.0,
         ]);
 
-        $queueForExecutions
-            ->setExecution($execution)
-            ->setProject($project)
-            ->trigger();
+        $bus->dispatch(new ExecutionCompleted(
+            execution: $execution->getArrayCopy(),
+            project: $project->getArrayCopy(),
+        ));
     }
 
     /**
@@ -320,7 +315,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
@@ -343,9 +337,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,
@@ -373,19 +366,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;
         }
 
@@ -592,26 +585,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);