Merge branch '1.6.x' into add-file-transformation-endpoints

2026-05-26 13:51:13 +00:00 · 2025-03-04 07:34:43 +00:00
parent 9849e7b232 eff58459ef
commit 239bebfe2e
21 changed files with 304 additions and 1063 deletions
@@ -88,9 +88,7 @@ RUN chmod +x /usr/local/bin/doctor && \
    chmod +x /usr/local/bin/worker-stats-usage && \
    chmod +x /usr/local/bin/worker-stats-usage-dump && \
    chmod +x /usr/local/bin/stats-resources && \
-    chmod +x /usr/local/bin/worker-stats-resources && \
-    chmod +x /usr/local/bin/worker-usage && \
-    chmod +x /usr/local/bin/worker-usage-dump
+    chmod +x /usr/local/bin/worker-stats-resources

 # Letsencrypt Permissions
 RUN mkdir -p /etc/letsencrypt/live/ && chmod -Rf 755 /etc/letsencrypt/live/
@@ -6,13 +6,14 @@ use Appwrite\Event\Build;
 use Appwrite\Event\Delete;
 use Appwrite\Event\Event;
 use Appwrite\Event\Func;
+use Appwrite\Event\Realtime;
 use Appwrite\Event\StatsUsage;
 use Appwrite\Event\Validator\FunctionEvent;
+use Appwrite\Event\Webhook;
 use Appwrite\Extend\Exception;
 use Appwrite\Extend\Exception as AppwriteException;
 use Appwrite\Functions\Validator\Headers;
 use Appwrite\Functions\Validator\RuntimeSpecification;
-use Appwrite\Messaging\Adapter\Realtime;
 use Appwrite\Platform\Tasks\ScheduleExecutions;
 use Appwrite\SDK\AuthType;
 use Appwrite\SDK\ContentType;
@@ -194,9 +195,12 @@ App::post('/v1/functions')
    ->inject('user')
    ->inject('queueForEvents')
    ->inject('queueForBuilds')
+    ->inject('queueForWebhooks')
+    ->inject('queueForFunctions')
+    ->inject('queueForRealtime')
    ->inject('dbForPlatform')
    ->inject('gitHub')
-    ->action(function (string $functionId, string $name, string $runtime, array $execute, array $events, string $schedule, int $timeout, bool $enabled, bool $logging, string $entrypoint, string $commands, array $scopes, string $installationId, string $providerRepositoryId, string $providerBranch, bool $providerSilentMode, string $providerRootDirectory, string $templateRepository, string $templateOwner, string $templateRootDirectory, string $templateVersion, string $specification, Request $request, Response $response, Database $dbForProject, callable $timelimit, Document $project, Document $user, Event $queueForEvents, Build $queueForBuilds, Database $dbForPlatform, GitHub $github) use ($redeployVcs) {
+    ->action(function (string $functionId, string $name, string $runtime, array $execute, array $events, string $schedule, int $timeout, bool $enabled, bool $logging, string $entrypoint, string $commands, array $scopes, string $installationId, string $providerRepositoryId, string $providerBranch, bool $providerSilentMode, string $providerRootDirectory, string $templateRepository, string $templateOwner, string $templateRootDirectory, string $templateVersion, string $specification, Request $request, Response $response, Database $dbForProject, callable $timelimit, Document $project, Document $user, Event $queueForEvents, Build $queueForBuilds, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Database $dbForPlatform, GitHub $github) use ($redeployVcs) {
        $functionId = ($functionId == 'unique()') ? ID::unique() : $functionId;

        // Temporary abuse check
@@ -386,51 +390,29 @@ App::post('/v1/functions')
                ]))
            );

-            /** Trigger Webhook */
            $ruleModel = new Rule();
            $ruleCreate =
                $queueForEvents
-                ->setClass(Event::WEBHOOK_CLASS_NAME)
-                ->setQueue(Event::WEBHOOK_QUEUE_NAME);
+                    ->setProject($project)
+                    ->setEvent('rules.[ruleId].create')
+                    ->setParam('ruleId', $rule->getId())
+                    ->setPayload($rule->getArrayCopy(array_keys($ruleModel->getRules())));

-            $ruleCreate
-                ->setProject($project)
-                ->setEvent('rules.[ruleId].create')
-                ->setParam('ruleId', $rule->getId())
-                ->setPayload($rule->getArrayCopy(array_keys($ruleModel->getRules())))
+            /** Trigger Webhook */
+            $queueForWebhooks
+                ->from($ruleCreate)
                ->trigger();

            /** Trigger Functions */
-            $ruleCreate
-                ->setClass(Event::FUNCTIONS_CLASS_NAME)
-                ->setQueue(Event::FUNCTIONS_QUEUE_NAME)
+            $queueForFunctions
+                ->from($ruleCreate)
                ->trigger();

-            /** Trigger realtime event */
-            $allEvents = Event::generateEvents('rules.[ruleId].create', [
-                'ruleId' => $rule->getId(),
-            ]);
-
-            $target = Realtime::fromPayload(
-                // Pass first, most verbose event pattern
-                event: $allEvents[0],
-                payload: $rule,
-                project: $project
-            );
-            Realtime::send(
-                projectId: 'console',
-                payload: $rule->getArrayCopy(),
-                events: $allEvents,
-                channels: $target['channels'],
-                roles: $target['roles']
-            );
-            Realtime::send(
-                projectId: $project->getId(),
-                payload: $rule->getArrayCopy(),
-                events: $allEvents,
-                channels: $target['channels'],
-                roles: $target['roles']
-            );
+            /** Trigger Realtime Events */
+            $queueForRealtime
+                ->from($ruleCreate)
+                ->setSubscribers(['console', $project->getId()])
+                ->trigger();
        }

        $queueForEvents->setParam('functionId', $function->getId());
@@ -6,7 +6,6 @@ use Appwrite\Auth\Auth;
 use Appwrite\ClamAV\Network;
 use Appwrite\Event\Delete;
 use Appwrite\Event\Event;
-use Appwrite\Event\StatsUsage;
 use Appwrite\Extend\Exception;
 use Appwrite\OpenSSL\OpenSSL;
 use Appwrite\SDK\AuthType;
@@ -939,8 +938,7 @@ App::get('/v1/storage/buckets/:bucketId/files/:fileId/preview')
    ->inject('dbForProject')
    ->inject('deviceForFiles')
    ->inject('deviceForLocal')
-    ->inject('queueForStatsUsage')
-    ->action(function (string $bucketId, string $fileId, int $width, int $height, string $gravity, int $quality, int $borderWidth, string $borderColor, int $borderRadius, float $opacity, int $rotation, string $background, string $output, Response $response, Database $dbForProject, Device $deviceForFiles, Device $deviceForLocal, StatsUsage $queueForStatsUsage) {
+    ->action(function (string $bucketId, string $fileId, int $width, int $height, string $gravity, int $quality, int $borderWidth, string $borderColor, int $borderRadius, float $opacity, int $rotation, string $background, string $output, Response $response, Database $dbForProject, Device $deviceForFiles, Device $deviceForLocal) {

        if (!\extension_loaded('imagick')) {
            throw new Exception(Exception::GENERAL_SERVER_ERROR, 'Imagick extension is missing');
@@ -1068,14 +1066,13 @@ App::get('/v1/storage/buckets/:bucketId/files/:fileId/preview')

        $contentType = (\array_key_exists($output, $outputs)) ? $outputs[$output] : $outputs['jpg'];

-        $queueForStatsUsage
-            ->addMetric(METRIC_FILES_TRANSFORMATIONS, 1)
-            ->addMetric(str_replace('{bucketInternalId}', $bucket->getInternalId(), METRIC_BUCKET_ID_FILES_TRANSFORMATIONS), 1);
-
-        $transformedAt = $file->getAttribute('transformedAt', '');
-        if (DateTime::formatTz(DateTime::addSeconds(new \DateTime(), -APP_PROJECT_ACCESS)) > $transformedAt) {
-            $file->setAttribute('transformedAt', DateTime::now());
-            Authorization::skip(fn () => $dbForProject->updateDocument('bucket_' .  $file->getAttribute('bucketInternalId'), $file->getId(), $file));
+        //Do not update transformedAt if it's a console user
+        if (!Auth::isPrivilegedUser(Authorization::getRoles())) {
+            $transformedAt = $file->getAttribute('transformedAt', '');
+            if (DateTime::formatTz(DateTime::addSeconds(new \DateTime(), -APP_PROJECT_ACCESS)) > $transformedAt) {
+                $file->setAttribute('transformedAt', DateTime::now());
+                Authorization::skip(fn () => $dbForProject->updateDocument('bucket_' . $file->getAttribute('bucketInternalId'), $file->getId(), $file));
+            }
        }

        $response
@@ -534,7 +534,7 @@ App::init()

                if ($type === 'bucket') {
                    $bucketId = $parts[1] ?? null;
-                    $bucket   = Authorization::skip(fn () => $dbForProject->getDocument('buckets', $bucketId));
+                    $bucket = Authorization::skip(fn () => $dbForProject->getDocument('buckets', $bucketId));

                    if ($bucket->isEmpty() || (!$bucket->getAttribute('enabled') && !$isAppUser && !$isPrivilegedUser)) {
                        throw new Exception(Exception::STORAGE_BUCKET_NOT_FOUND);
@@ -560,11 +560,13 @@ App::init()
                    if ($file->isEmpty()) {
                        throw new Exception(Exception::STORAGE_FILE_NOT_FOUND);
                    }
-
-                    $transformedAt = $file->getAttribute('transformedAt', '');
-                    if (DateTime::formatTz(DateTime::addSeconds(new \DateTime(), -APP_PROJECT_ACCESS)) > $transformedAt) {
-                        $file->setAttribute('transformedAt', DateTime::now());
-                        Authorization::skip(fn () => $dbForProject->updateDocument('bucket_' . $file->getAttribute('bucketInternalId'), $file->getId(), $file));
+                    //Do not update transformedAt if it's a console user
+                    if (!Auth::isPrivilegedUser(Authorization::getRoles())) {
+                        $transformedAt = $file->getAttribute('transformedAt', '');
+                        if (DateTime::formatTz(DateTime::addSeconds(new \DateTime(), -APP_PROJECT_ACCESS)) > $transformedAt) {
+                            $file->setAttribute('transformedAt', DateTime::now());
+                            Authorization::skip(fn () => $dbForProject->updateDocument('bucket_' . $file->getAttribute('bucketInternalId'), $file->getId(), $file));
+                        }
                    }
                }

@@ -1200,9 +1200,6 @@ App::setResource('queueForAudits', function (Queue\Publisher $publisher) {
 App::setResource('queueForFunctions', function (Queue\Publisher $publisher) {
    return new Func($publisher);
 }, ['publisher']);
-App::setResource('queueForUsage', function (Queue\Publisher $publisher) {
-    return new Usage($publisher);
-}, ['publisher']);
 App::setResource('queueForCertificates', function (Queue\Publisher $publisher) {
    return new Certificate($publisher);
 }, ['publisher']);
@@ -13,12 +13,12 @@ use Appwrite\Event\Func;
 use Appwrite\Event\Mail;
 use Appwrite\Event\Messaging;
 use Appwrite\Event\Migration;
+use Appwrite\Event\Realtime;
 use Appwrite\Event\StatsUsage;
 use Appwrite\Event\StatsUsageDump;
 /** remove */
-use Appwrite\Event\Usage;
-use Appwrite\Event\UsageDump;
 /** /remove */
+use Appwrite\Event\Webhook;
 use Appwrite\Platform\Appwrite;
 use Swoole\Runtime;
 use Utopia\Abuse\Adapters\TimeLimit\Redis as TimeLimitRedis;
@@ -269,14 +269,6 @@ Server::setResource('consumer', function (Group $pools) {
    return $pools->get('consumer')->pop()->getResource();
 }, ['pools']);

-Server::setResource('queueForUsage', function (Publisher $publisher) {
-    return new Usage($publisher);
-}, ['publisher']);
-
-Server::setResource('queueForUsageDump', function (Publisher $publisher) {
-    return new UsageDump($publisher);
-}, ['publisher']);
-
 Server::setResource('queueForStatsUsage', function (Publisher $publisher) {
    return new StatsUsage($publisher);
 }, ['publisher']);
@@ -313,10 +305,18 @@ Server::setResource('queueForAudits', function (Publisher $publisher) {
    return new Audit($publisher);
 }, ['publisher']);

+Server::setResource('queueForWebhooks', function (Publisher $publisher) {
+    return new Webhook($publisher);
+}, ['publisher']);
+
 Server::setResource('queueForFunctions', function (Publisher $publisher) {
    return new Func($publisher);
 }, ['publisher']);

+Server::setResource('queueForRealtime', function () {
+    return new Realtime();
+}, []);
+
 Server::setResource('queueForCertificates', function (Publisher $publisher) {
    return new Certificate($publisher);
 }, ['publisher']);
@@ -1,3 +0,0 @@
-#!/bin/sh
-
-php /usr/src/code/app/worker.php usage $@
@@ -1,3 +0,0 @@
-#!/bin/sh
-
-php /usr/src/code/app/worker.php usage-dump $@
@@ -7,10 +7,17 @@ use Utopia\Database\Document;

 class Realtime extends Event
 {
+    protected array $subscribers = [];
+
    public function __construct()
    {
    }

+    /**
+     * Get Realtime payload for this event.
+     *
+     * @return array
+     */
    public function getRealtimePayload(): array
    {
        $payload = [];
@@ -24,6 +31,28 @@ class Realtime extends Event
        return $payload;
    }

+    /**
+     * Set subscribers for this realtime event.
+     *
+     * @param array $subscribers
+     * @return array
+     */
+    public function setSubscribers(array $subscribers): self
+    {
+        $this->subscribers = $subscribers;
+        return $this;
+    }
+
+    /**
+     * Get subscribers for this realtime event.
+     *
+     * @return array
+     */
+    public function getSubscribers(): array
+    {
+        return $this->subscribers;
+    }
+
    /**
     * Execute Event.
     *
@@ -53,17 +82,23 @@ class Realtime extends Event
            bucket: $bucket,
        );

-        RealtimeAdapter::send(
-            projectId: $target['projectId'] ?? $this->getProject()->getId(),
-            payload: $this->getRealtimePayload(),
-            events: $allEvents,
-            channels: $target['channels'],
-            roles: $target['roles'],
-            options: [
-                'permissionsChanged' => $target['permissionsChanged'],
-                'userId' => $this->getParam('userId')
-            ]
-        );
+        $projectIds = !empty($this->getSubscribers())
+            ? $this->getSubscribers()
+            : [$target['projectId'] ?? $this->getProject()->getId()];
+
+        foreach ($projectIds as $projectId) {
+            RealtimeAdapter::send(
+                projectId: $projectId,
+                payload: $this->getRealtimePayload(),
+                events: $allEvents,
+                channels: $target['channels'],
+                roles: $target['roles'],
+                options: [
+                    'permissionsChanged' => $target['permissionsChanged'],
+                    'userId' => $this->getParam('userId')
+                ]
+            );
+        }

        return true;
    }
@@ -1,78 +0,0 @@
-<?php
-
-namespace Appwrite\Event;
-
-use Utopia\Database\Document;
-use Utopia\Queue\Publisher;
-
-class Usage extends Event
-{
-    protected array $metrics = [];
-    protected array $reduce  = [];
-
-    public function __construct(protected Publisher $publisher)
-    {
-        parent::__construct($publisher);
-
-        $this
-            ->setQueue(Event::USAGE_QUEUE_NAME)
-            ->setClass(Event::USAGE_CLASS_NAME);
-    }
-
-    /**
-     * Add reduce.
-     *
-     * @param Document $document
-     * @return self
-     */
-    public function addReduce(Document $document): self
-    {
-        $this->reduce[] = $document;
-
-        return $this;
-    }
-
-    /**
-     * Add metric.
-     *
-     * @param string $key
-     * @param int $value
-     * @return self
-     */
-    public function addMetric(string $key, int $value): self
-    {
-
-        $this->metrics[] = [
-            'key' => $key,
-            'value' => $value,
-        ];
-
-        return $this;
-    }
-
-    /**
-     * Prepare the payload for the usage event.
-     *
-     * @return array
-     */
-    protected function preparePayload(): array
-    {
-        return [
-            'project' => $this->project,
-            'reduce'  => $this->reduce,
-            'metrics' => $this->metrics,
-        ];
-    }
-
-    /**
-     * Sends metrics to the usage worker.
-     *
-     * @return string|bool
-     */
-    public function trigger(): string|bool
-    {
-        parent::trigger();
-        $this->metrics = [];
-        return true;
-    }
-}
@@ -1,44 +0,0 @@
-<?php
-
-namespace Appwrite\Event;
-
-use Utopia\Queue\Publisher;
-
-class UsageDump extends Event
-{
-    protected array $stats;
-
-    public function __construct(protected Publisher $publisher)
-    {
-        parent::__construct($publisher);
-
-        $this
-            ->setQueue(Event::USAGE_DUMP_QUEUE_NAME)
-            ->setClass(Event::USAGE_DUMP_CLASS_NAME);
-    }
-
-    /**
-     * Add Stats.
-     *
-     * @param array $stats
-     * @return self
-     */
-    public function setStats(array $stats): self
-    {
-        $this->stats = $stats;
-
-        return $this;
-    }
-
-    /**
-     * Prepare the payload for the usage dump event.
-     *
-     * @return array
-     */
-    protected function preparePayload(): array
-    {
-        return [
-            'stats' => $this->stats,
-        ];
-    }
-}
@@ -14,10 +14,6 @@ use Appwrite\Platform\Workers\Migrations;
 use Appwrite\Platform\Workers\StatsResources;
 use Appwrite\Platform\Workers\StatsUsage;
 use Appwrite\Platform\Workers\StatsUsageDump;
-/** remove */
-use Appwrite\Platform\Workers\Usage;
-use Appwrite\Platform\Workers\UsageDump;
-/** /remove */
 use Appwrite\Platform\Workers\Webhooks;
 use Utopia\Platform\Service;

@@ -40,10 +36,6 @@ class Workers extends Service
            ->addAction(StatsUsage::getName(), new StatsUsage())
            ->addAction(Migrations::getName(), new Migrations())
            ->addAction(StatsResources::getName(), new StatsResources())
-            /** Remove */
-            ->addAction(UsageDump::getName(), new UsageDump())
-            ->addAction(Usage::getName(), new Usage())
-            /** /remove */
        ;
    }
 }
@@ -62,7 +62,7 @@ class StatsResources extends Action
            Authorization::disable();
            Authorization::setDefaultStatus(false);

-            $last24Hours = (new \DateTime())->sub(\DateInterval::createFromDateString('3 hours'));
+            $last24Hours = (new \DateTime())->sub(\DateInterval::createFromDateString('24 hours'));
            /**
             * For each project that were accessed in last 24 hours
             */
@@ -5,8 +5,9 @@ namespace Appwrite\Platform\Workers;
 use Ahc\Jwt\JWT;
 use Appwrite\Event\Event;
 use Appwrite\Event\Func;
+use Appwrite\Event\Realtime;
 use Appwrite\Event\StatsUsage;
-use Appwrite\Messaging\Adapter\Realtime;
+use Appwrite\Event\Webhook;
 use Appwrite\Utopia\Response\Model\Deployment;
 use Appwrite\Vcs\Comment;
 use Exception;
@@ -49,15 +50,17 @@ class Builds extends Action
            ->inject('project')
            ->inject('dbForPlatform')
            ->inject('queueForEvents')
+            ->inject('queueForWebhooks')
            ->inject('queueForFunctions')
+            ->inject('queueForRealtime')
            ->inject('queueForStatsUsage')
            ->inject('cache')
            ->inject('dbForProject')
            ->inject('deviceForFunctions')
            ->inject('isResourceBlocked')
            ->inject('log')
-            ->callback(fn ($message, Document $project, Database $dbForPlatform, Event $queueForEvents, Func $queueForFunctions, StatsUsage $usage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, callable $isResourceBlocked, Log $log) =>
-                $this->action($message, $project, $dbForPlatform, $queueForEvents, $queueForFunctions, $usage, $cache, $dbForProject, $deviceForFunctions, $isResourceBlocked, $log));
+            ->callback(fn ($message, Document $project, Database $dbForPlatform, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, StatsUsage $usage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, callable $isResourceBlocked, Log $log) =>
+                $this->action($message, $project, $dbForPlatform, $queueForEvents, $queueForWebhooks, $queueForFunctions, $queueForRealtime, $usage, $cache, $dbForProject, $deviceForFunctions, $isResourceBlocked, $log));
    }

    /**
@@ -65,7 +68,9 @@ class Builds extends Action
     * @param Document $project
     * @param Database $dbForPlatform
     * @param Event $queueForEvents
+     * @param Webhook $queueForWebhooks
     * @param Func $queueForFunctions
+     * @param Realtime $queueForRealtime
     * @param StatsUsage $queueForStatsUsage
     * @param Cache $cache
     * @param Database $dbForProject
@@ -74,7 +79,7 @@ class Builds extends Action
     * @return void
     * @throws \Utopia\Database\Exception
     */
-    public function action(Message $message, Document $project, Database $dbForPlatform, Event $queueForEvents, Func $queueForFunctions, StatsUsage $queueForStatsUsage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, callable $isResourceBlocked, Log $log): void
+    public function action(Message $message, Document $project, Database $dbForPlatform, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, StatsUsage $queueForStatsUsage, Cache $cache, Database $dbForProject, Device $deviceForFunctions, callable $isResourceBlocked, Log $log): void
    {
        $payload = $message->getPayload() ?? [];

@@ -95,7 +100,7 @@ class Builds extends Action
            case BUILD_TYPE_RETRY:
                Console::info('Creating build for deployment: ' . $deployment->getId());
                $github = new GitHub($cache);
-                $this->buildDeployment($deviceForFunctions, $queueForFunctions, $queueForEvents, $queueForStatsUsage, $dbForPlatform, $dbForProject, $github, $project, $resource, $deployment, $template, $isResourceBlocked, $log);
+                $this->buildDeployment($deviceForFunctions, $queueForWebhooks, $queueForFunctions, $queueForRealtime, $queueForEvents, $queueForStatsUsage, $dbForPlatform, $dbForProject, $github, $project, $resource, $deployment, $template, $isResourceBlocked, $log);
                break;

            default:
@@ -105,7 +110,9 @@ class Builds extends Action

    /**
     * @param Device $deviceForFunctions
+     * @param Webhook $queueForWebhooks
     * @param Func $queueForFunctions
+     * @param Realtime $queueForRealtime
     * @param Event $queueForEvents
     * @param StatsUsage $queueForStatsUsage
     * @param Database $dbForPlatform
@@ -120,7 +127,7 @@ class Builds extends Action
     * @throws \Utopia\Database\Exception
     * @throws Exception
     */
-    protected function buildDeployment(Device $deviceForFunctions, Func $queueForFunctions, Event $queueForEvents, StatsUsage $queueForStatsUsage, Database $dbForPlatform, Database $dbForProject, GitHub $github, Document $project, Document $function, Document $deployment, Document $template, callable $isResourceBlocked, Log $log): void
+    protected function buildDeployment(Device $deviceForFunctions, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Event $queueForEvents, StatsUsage $queueForStatsUsage, Database $dbForPlatform, Database $dbForProject, GitHub $github, Document $project, Document $function, Document $deployment, Document $template, callable $isResourceBlocked, Log $log): void
    {
        $executor = new Executor(System::getEnv('_APP_EXECUTOR_HOST'));

@@ -158,10 +165,7 @@ class Builds extends Action
        }

        // Realtime preparation
-        $allEvents = Event::generateEvents('functions.[functionId].deployments.[deploymentId].update', [
-            'functionId' => $function->getId(),
-            'deploymentId' => $deployment->getId()
-        ]);
+        $event = "functions.[functionId].deployments.[deploymentId].update";

        $startTime = DateTime::now();
        $durationStart = \microtime(true);
@@ -375,21 +379,16 @@ class Builds extends Action
                    $deployment = $dbForProject->updateDocument('deployments', $deployment->getId(), $deployment);

                    /**
-                     * Send realtime Event
+                     * Trigger Realtime Event
                     */
-                    $target = Realtime::fromPayload(
-                        // Pass first, most verbose event pattern
-                        event: $allEvents[0],
-                        payload: $build,
-                        project: $project
-                    );
-                    Realtime::send(
-                        projectId: 'console',
-                        payload: $build->getArrayCopy(),
-                        events: $allEvents,
-                        channels: $target['channels'],
-                        roles: $target['roles']
-                    );
+                    $queueForRealtime
+                        ->setProject($project)
+                        ->setSubscribers(['console'])
+                        ->setEvent($event)
+                        ->setParam('functionId', $function->getId())
+                        ->setParam('deploymentId', $deployment->getId())
+                        ->setPayload($build->getArrayCopy())
+                        ->trigger();
                }

                $tmpPath = '/tmp/builds/' . $buildId;
@@ -436,40 +435,34 @@ class Builds extends Action
                $this->runGitAction('building', $github, $providerCommitHash, $owner, $repositoryName, $project, $function, $deployment->getId(), $dbForProject, $dbForPlatform);
            }

-            /** Trigger Webhook */
            $deploymentModel = new Deployment();
            $deploymentUpdate =
                $queueForEvents
-                    ->setQueue(Event::WEBHOOK_QUEUE_NAME)
-                    ->setClass(Event::WEBHOOK_CLASS_NAME)
                    ->setProject($project)
                    ->setEvent('functions.[functionId].deployments.[deploymentId].update')
                    ->setParam('functionId', $function->getId())
                    ->setParam('deploymentId', $deployment->getId())
                    ->setPayload($deployment->getArrayCopy(array_keys($deploymentModel->getRules())));

-            $deploymentUpdate->trigger();
+            /** Trigger Webhook */
+            $queueForWebhooks
+                ->from($deploymentUpdate)
+                ->trigger();

            /** Trigger Functions */
            $queueForFunctions
                ->from($deploymentUpdate)
                ->trigger();

-            /** Trigger Realtime */
-            $target = Realtime::fromPayload(
-                // Pass first, most verbose event pattern
-                event: $allEvents[0],
-                payload: $build,
-                project: $project
-            );
-
-            Realtime::send(
-                projectId: 'console',
-                payload: $build->getArrayCopy(),
-                events: $allEvents,
-                channels: $target['channels'],
-                roles: $target['roles']
-            );
+            /** Trigger Realtime Event */
+            $queueForRealtime
+                ->setProject($project)
+                ->setSubscribers(['console'])
+                ->setEvent($event)
+                ->setParam('functionId', $function->getId())
+                ->setParam('deploymentId', $deployment->getId())
+                ->setPayload($build->getArrayCopy())
+                ->trigger();

            $vars = [];

@@ -562,12 +555,12 @@ class Builds extends Action
                        $err = $error;
                    }
                }),
-                Co\go(function () use ($executor, $project, $deployment, &$response, &$build, $dbForProject, $allEvents, &$err, &$isCanceled) {
+                Co\go(function () use ($executor, $project, $function, $deployment, &$response, &$build, $dbForProject, $event, &$err, $queueForRealtime, &$isCanceled) {
                    try {
                        $executor->getLogs(
                            deploymentId: $deployment->getId(),
                            projectId: $project->getId(),
-                            callback: function ($logs) use (&$response, &$err, &$build, $dbForProject, $allEvents, $project, &$isCanceled) {
+                            callback: function ($logs) use (&$response, &$err, &$build, $dbForProject, $event, $project, $function, $deployment, $queueForRealtime, &$isCanceled) {
                                if ($isCanceled) {
                                    return;
                                }
@@ -592,21 +585,16 @@ class Builds extends Action
                                    $build = $dbForProject->updateDocument('builds', $build->getId(), $build);

                                    /**
-                                     * Send realtime Event
+                                     * Trigger Realtime Event
                                     */
-                                    $target = Realtime::fromPayload(
-                                        // Pass first, most verbose event pattern
-                                        event: $allEvents[0],
-                                        payload: $build,
-                                        project: $project
-                                    );
-                                    Realtime::send(
-                                        projectId: 'console',
-                                        payload: $build->getArrayCopy(),
-                                        events: $allEvents,
-                                        channels: $target['channels'],
-                                        roles: $target['roles']
-                                    );
+                                    $queueForRealtime
+                                        ->setProject($project)
+                                        ->setSubscribers(['console'])
+                                        ->setEvent($event)
+                                        ->setParam('functionId', $function->getId())
+                                        ->setParam('deploymentId', $deployment->getId())
+                                        ->setPayload($build->getArrayCopy())
+                                        ->trigger();
                                }
                            }
                        );
@@ -694,21 +682,16 @@ class Builds extends Action
            }
        } finally {
            /**
-             * Send realtime Event
+             * Trigger Realtime Event
             */
-            $target = Realtime::fromPayload(
-                // Pass first, most verbose event pattern
-                event: $allEvents[0],
-                payload: $build,
-                project: $project
-            );
-            Realtime::send(
-                projectId: 'console',
-                payload: $build->getArrayCopy(),
-                events: $allEvents,
-                channels: $target['channels'],
-                roles: $target['roles']
-            );
+            $queueForRealtime
+                ->setProject($project)
+                ->setSubscribers(['console'])
+                ->setEvent($event)
+                ->setParam('functionId', $function->getId())
+                ->setParam('deploymentId', $deployment->getId())
+                ->setPayload($build->getArrayCopy())
+                ->trigger();

            /** Trigger usage queue */
            if ($build->getAttribute('status') === 'ready') {
@@ -6,7 +6,8 @@ use Appwrite\Certificates\Adapter as CertificatesAdapter;
 use Appwrite\Event\Event;
 use Appwrite\Event\Func;
 use Appwrite\Event\Mail;
-use Appwrite\Messaging\Adapter\Realtime;
+use Appwrite\Event\Realtime;
+use Appwrite\Event\Webhook;
 use Appwrite\Network\Validator\CNAME;
 use Appwrite\Template\Template;
 use Appwrite\Utopia\Response\Model\Rule;
@@ -46,12 +47,14 @@ class Certificates extends Action
            ->inject('dbForPlatform')
            ->inject('queueForMails')
            ->inject('queueForEvents')
+            ->inject('queueForWebhooks')
            ->inject('queueForFunctions')
+            ->inject('queueForRealtime')
            ->inject('log')
            ->inject('certificates')
            ->callback(
-                fn (Message $message, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Func $queueForFunctions, Log $log, CertificatesAdapter $certificates) =>
-                    $this->action($message, $dbForPlatform, $queueForMails, $queueForEvents, $queueForFunctions, $log, $certificates)
+                fn (Message $message, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Log $log, CertificatesAdapter $certificates) =>
+                    $this->action($message, $dbForPlatform, $queueForMails, $queueForEvents, $queueForWebhooks, $queueForFunctions, $queueForRealtime, $log, $certificates)
            );
    }

@@ -60,14 +63,16 @@ class Certificates extends Action
     * @param Database $dbForPlatform
     * @param Mail $queueForMails
     * @param Event $queueForEvents
+     * @param Webhook $queueForWebhooks
     * @param Func $queueForFunctions
+     * @param Realtime $queueForRealtime
     * @param Log $log
     * @param CertificatesAdapter $certificates
     * @return void
     * @throws Throwable
     * @throws \Utopia\Database\Exception
     */
-    public function action(Message $message, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Func $queueForFunctions, Log $log, CertificatesAdapter $certificates): void
+    public function action(Message $message, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Log $log, CertificatesAdapter $certificates): void
    {
        $payload = $message->getPayload() ?? [];

@@ -81,7 +86,7 @@ class Certificates extends Action

        $log->addTag('domain', $domain->get());

-        $this->execute($domain, $dbForPlatform, $queueForMails, $queueForEvents, $queueForFunctions, $log, $certificates, $skipRenewCheck);
+        $this->execute($domain, $dbForPlatform, $queueForMails, $queueForEvents, $queueForWebhooks, $queueForFunctions, $queueForRealtime, $log, $certificates, $skipRenewCheck);
    }

    /**
@@ -90,13 +95,14 @@ class Certificates extends Action
     * @param Mail $queueForMails
     * @param Event $queueForEvents
     * @param Func $queueForFunctions
+     * @param Realtime $queueForRealtime
     * @param CertificatesAdapter $certificates
     * @param bool $skipRenewCheck
     * @return void
     * @throws Throwable
     * @throws \Utopia\Database\Exception
     */
-    private function execute(Domain $domain, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Func $queueForFunctions, Log $log, CertificatesAdapter $certificates, bool $skipRenewCheck = false): void
+    private function execute(Domain $domain, Database $dbForPlatform, Mail $queueForMails, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Log $log, CertificatesAdapter $certificates, bool $skipRenewCheck = false): void
    {
        /**
         * 1. Read arguments and validate domain
@@ -186,7 +192,7 @@ class Certificates extends Action
            $certificate->setAttribute('updated', DateTime::now());

            // Save all changes we made to certificate document into database
-            $this->saveCertificateDocument($domain->get(), $certificate, $success, $dbForPlatform, $queueForEvents, $queueForFunctions);
+            $this->saveCertificateDocument($domain->get(), $certificate, $success, $dbForPlatform, $queueForEvents, $queueForWebhooks, $queueForFunctions, $queueForRealtime);
        }
    }

@@ -199,13 +205,14 @@ class Certificates extends Action
     * @param Database $dbForPlatform Database connection for console
     * @param Event $queueForEvents
     * @param Func $queueForFunctions
+     * @param Realtime $queueForRealtime
     * @return void
     * @throws \Utopia\Database\Exception
     * @throws Authorization
     * @throws Conflict
     * @throws Structure
     */
-    private function saveCertificateDocument(string $domain, Document $certificate, bool $success, Database $dbForPlatform, Event $queueForEvents, Func $queueForFunctions): void
+    private function saveCertificateDocument(string $domain, Document $certificate, bool $success, Database $dbForPlatform, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime): void
    {
        // Check if update or insert required
        $certificateDocument = $dbForPlatform->findOne('certificates', [Query::equal('domain', [$domain])]);
@@ -219,7 +226,7 @@ class Certificates extends Action
        }

        $certificateId = $certificate->getId();
-        $this->updateDomainDocuments($certificateId, $domain, $success, $dbForPlatform, $queueForEvents, $queueForFunctions);
+        $this->updateDomainDocuments($certificateId, $domain, $success, $dbForPlatform, $queueForEvents, $queueForWebhooks, $queueForFunctions, $queueForRealtime);
    }

    /**
@@ -338,7 +345,7 @@ class Certificates extends Action
     *
     * @return void
     */
-    private function updateDomainDocuments(string $certificateId, string $domain, bool $success, Database $dbForPlatform, Event $queueForEvents, Func $queueForFunctions): void
+    private function updateDomainDocuments(string $certificateId, string $domain, bool $success, Database $dbForPlatform, Event $queueForEvents, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime): void
    {
        // TODO: @christyjacob remove once we migrate the rules in 1.7.x
        if (System::getEnv('_APP_RULES_FORMAT') === 'md5') {
@@ -367,50 +374,28 @@ class Certificates extends Action
                return;
            }

-            /** Trigger Webhook */
            $ruleModel = new Rule();
            $queueForEvents
-                ->setQueue(Event::WEBHOOK_QUEUE_NAME)
-                ->setClass(Event::WEBHOOK_CLASS_NAME)
                ->setProject($project)
                ->setEvent('rules.[ruleId].update')
                ->setParam('ruleId', $rule->getId())
-                ->setPayload($rule->getArrayCopy(array_keys($ruleModel->getRules())))
-                ->trigger();
+                ->setPayload($rule->getArrayCopy(array_keys($ruleModel->getRules())));

+            /** Trigger Webhook */
+            $queueForWebhooks
+                ->from($queueForEvents)
+                ->trigger();

            /** Trigger Functions */
            $queueForFunctions
-                ->setProject($project)
-                ->setEvent('rules.[ruleId].update')
-                ->setParam('ruleId', $rule->getId())
-                ->setPayload($rule->getArrayCopy(array_keys($ruleModel->getRules())))
+                ->from($queueForEvents)
                ->trigger();

-            /** Trigger realtime event */
-            $allEvents = Event::generateEvents('rules.[ruleId].update', [
-                'ruleId' => $rule->getId(),
-            ]);
-            $target = Realtime::fromPayload(
-                // Pass first, most verbose event pattern
-                event: $allEvents[0],
-                payload: $rule,
-                project: $project
-            );
-            Realtime::send(
-                projectId: 'console',
-                payload: $rule->getArrayCopy(),
-                events: $allEvents,
-                channels: $target['channels'],
-                roles: $target['roles']
-            );
-            Realtime::send(
-                projectId: $project->getId(),
-                payload: $rule->getArrayCopy(),
-                events: $allEvents,
-                channels: $target['channels'],
-                roles: $target['roles']
-            );
+            /** Trigger Realtime Events */
+            $queueForRealtime
+                ->from($queueForEvents)
+                ->setSubscribers(['console', $projectId])
+                ->trigger();
        }
    }
 }
@@ -2,8 +2,7 @@

 namespace Appwrite\Platform\Workers;

-use Appwrite\Event\Event;
-use Appwrite\Messaging\Adapter\Realtime;
+use Appwrite\Event\Realtime;
 use Exception;
 use Utopia\CLI\Console;
 use Utopia\Database\Database;
@@ -37,8 +36,9 @@ class Databases extends Action
            ->inject('project')
            ->inject('dbForPlatform')
            ->inject('dbForProject')
+            ->inject('queueForRealtime')
            ->inject('log')
-            ->callback(fn (Message $message, Document $project, Database $dbForPlatform, Database $dbForProject, Log $log) => $this->action($message, $project, $dbForPlatform, $dbForProject, $log));
+            ->callback(fn (Message $message, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime, Log $log) => $this->action($message, $project, $dbForPlatform, $dbForProject, $queueForRealtime, $log));
    }

    /**
@@ -46,16 +46,17 @@ class Databases extends Action
     * @param Document $project
     * @param Database $dbForPlatform
     * @param Database $dbForProject
+     * @param Realtime $queueForRealtime
     * @param Log $log
     * @return void
     * @throws \Exception
     */
-    public function action(Message $message, Document $project, Database $dbForPlatform, Database $dbForProject, Log $log): void
+    public function action(Message $message, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime, Log $log): void
    {
        $payload = $message->getPayload() ?? [];

        if (empty($payload)) {
-            throw new \Exception('Missing payload');
+            throw new Exception('Missing payload');
        }

        $type = $payload['type'];
@@ -75,11 +76,11 @@ class Databases extends Action
        match (\strval($type)) {
            DATABASE_TYPE_DELETE_DATABASE => $this->deleteDatabase($database, $project, $dbForProject),
            DATABASE_TYPE_DELETE_COLLECTION => $this->deleteCollection($database, $collection, $project, $dbForProject),
-            DATABASE_TYPE_CREATE_ATTRIBUTE => $this->createAttribute($database, $collection, $document, $project, $dbForPlatform, $dbForProject),
-            DATABASE_TYPE_DELETE_ATTRIBUTE => $this->deleteAttribute($database, $collection, $document, $project, $dbForPlatform, $dbForProject),
-            DATABASE_TYPE_CREATE_INDEX => $this->createIndex($database, $collection, $document, $project, $dbForPlatform, $dbForProject),
-            DATABASE_TYPE_DELETE_INDEX => $this->deleteIndex($database, $collection, $document, $project, $dbForPlatform, $dbForProject),
-            default => throw new \Exception('No database operation for type: ' . \strval($type)),
+            DATABASE_TYPE_CREATE_ATTRIBUTE => $this->createAttribute($database, $collection, $document, $project, $dbForPlatform, $dbForProject, $queueForRealtime),
+            DATABASE_TYPE_DELETE_ATTRIBUTE => $this->deleteAttribute($database, $collection, $document, $project, $dbForPlatform, $dbForProject, $queueForRealtime),
+            DATABASE_TYPE_CREATE_INDEX => $this->createIndex($database, $collection, $document, $project, $dbForPlatform, $dbForProject, $queueForRealtime),
+            DATABASE_TYPE_DELETE_INDEX => $this->deleteIndex($database, $collection, $document, $project, $dbForPlatform, $dbForProject, $queueForRealtime),
+            default => throw new Exception('No database operation for type: ' . \strval($type)),
        };
    }

@@ -90,13 +91,14 @@ class Databases extends Action
     * @param Document $project
     * @param Database $dbForPlatform
     * @param Database $dbForProject
+     * @param Realtime $queueForRealtime
     * @return void
     * @throws Authorization
     * @throws Conflict
     * @throws \Exception
     * @throws \Throwable
     */
-    private function createAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForPlatform, Database $dbForProject): void
+    private function createAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime): void
    {
        if ($collection->isEmpty()) {
            throw new Exception('Missing collection');
@@ -106,12 +108,7 @@ class Databases extends Action
        }

        $projectId = $project->getId();
-
-        $events = Event::generateEvents('databases.[databaseId].collections.[collectionId].attributes.[attributeId].update', [
-            'databaseId' => $database->getId(),
-            'collectionId' => $collection->getId(),
-            'attributeId' => $attribute->getId()
-        ]);
+        $event = "databases.[databaseId].collections.[collectionId].attributes.[attributeId].update";
        /**
         * TODO @christyjacob4 verify if this is still the case
         * Fetch attribute from the database, since with Resque float values are loosing informations.
@@ -169,7 +166,7 @@ class Databases extends Action
                    break;
                default:
                    if (!$dbForProject->createAttribute('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId(), $key, $type, $size, $required, $default, $signed, $array, $format, $formatOptions, $filters)) {
-                        throw new \Exception('Failed to create Attribute');
+                        throw new Exception('Failed to create Attribute');
                    }
            }

@@ -200,7 +197,7 @@ class Databases extends Action

            throw $e;
        } finally {
-            $this->trigger($database, $collection, $attribute, $project, $projectId, $events);
+            $this->trigger($database, $collection, $project, $event, $queueForRealtime, $attribute);

            if (! $relatedCollection->isEmpty()) {
                $dbForProject->purgeCachedDocument('database_' . $database->getInternalId(), $relatedCollection->getId());
@@ -217,13 +214,14 @@ class Databases extends Action
     * @param Document $project
     * @param Database $dbForPlatform
     * @param Database $dbForProject
+     * @param Realtime $queueForRealtime
     * @return void
     * @throws Authorization
     * @throws Conflict
     * @throws \Exception
     * @throws \Throwable
     **/
-    private function deleteAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForPlatform, Database $dbForProject): void
+    private function deleteAttribute(Document $database, Document $collection, Document $attribute, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime): void
    {
        if ($collection->isEmpty()) {
            throw new Exception('Missing collection');
@@ -233,15 +231,9 @@ class Databases extends Action
        }

        $projectId = $project->getId();
-
-        $events = Event::generateEvents('databases.[databaseId].collections.[collectionId].attributes.[attributeId].delete', [
-            'databaseId' => $database->getId(),
-            'collectionId' => $collection->getId(),
-            'attributeId' => $attribute->getId()
-        ]);
+        $event = 'databases.[databaseId].collections.[collectionId].attributes.[attributeId].delete';
        $collectionId = $collection->getId();
        $key = $attribute->getAttribute('key', '');
-        $status = $attribute->getAttribute('status', '');
        $type = $attribute->getAttribute('type', '');
        $project = $dbForPlatform->getDocument('projects', $projectId);
        $options = $attribute->getAttribute('options', []);
@@ -312,7 +304,7 @@ class Databases extends Action

                throw $e;
            } finally {
-                $this->trigger($database, $collection, $attribute, $project, $projectId, $events);
+                $this->trigger($database, $collection, $project, $event, $queueForRealtime, $attribute);
            }

            // The underlying database removes/rebuilds indexes when attribute is removed
@@ -358,7 +350,7 @@ class Databases extends Action
                        }

                        if ($exists) { // Delete the duplicate if created, else update in db
-                            $this->deleteIndex($database, $collection, $index, $project, $dbForPlatform, $dbForProject);
+                            $this->deleteIndex($database, $collection, $index, $project, $dbForPlatform, $dbForProject, $queueForRealtime);
                        } else {
                            $dbForProject->updateDocument('indexes', $index->getId(), $index);
                        }
@@ -381,6 +373,7 @@ class Databases extends Action
     * @param Document $project
     * @param Database $dbForPlatform
     * @param Database $dbForProject
+     * @param Realtime $queueForRealtime
     * @return void
     * @throws Authorization
     * @throws Conflict
@@ -388,7 +381,7 @@ class Databases extends Action
     * @throws DatabaseException
     * @throws \Throwable
     */
-    private function createIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForPlatform, Database $dbForProject): void
+    private function createIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime): void
    {
        if ($collection->isEmpty()) {
            throw new Exception('Missing collection');
@@ -398,12 +391,7 @@ class Databases extends Action
        }

        $projectId = $project->getId();
-
-        $events = Event::generateEvents('databases.[databaseId].collections.[collectionId].indexes.[indexId].update', [
-            'databaseId' => $database->getId(),
-            'collectionId' => $collection->getId(),
-            'indexId' => $index->getId()
-        ]);
+        $event = 'databases.[databaseId].collections.[collectionId].indexes.[indexId].update';
        $collectionId = $collection->getId();
        $key = $index->getAttribute('key', '');
        $type = $index->getAttribute('type', '');
@@ -430,7 +418,7 @@ class Databases extends Action

            throw $e;
        } finally {
-            $this->trigger($database, $collection, $index, $project, $projectId, $events);
+            $this->trigger($database, $collection, $project, $event, $queueForRealtime, null, $index);
            $dbForProject->purgeCachedDocument('database_' . $database->getInternalId(), $collectionId);
        }
    }
@@ -442,6 +430,7 @@ class Databases extends Action
     * @param Document $project
     * @param Database $dbForPlatform
     * @param Database $dbForProject
+     * @param Realtime $queueForRealtime
     * @return void
     * @throws Authorization
     * @throws Conflict
@@ -449,7 +438,7 @@ class Databases extends Action
     * @throws DatabaseException
     * @throws \Throwable
     */
-    private function deleteIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForPlatform, Database $dbForProject): void
+    private function deleteIndex(Document $database, Document $collection, Document $index, Document $project, Database $dbForPlatform, Database $dbForProject, Realtime $queueForRealtime): void
    {
        if ($collection->isEmpty()) {
            throw new Exception('Missing collection');
@@ -459,12 +448,7 @@ class Databases extends Action
        }

        $projectId = $project->getId();
-
-        $events = Event::generateEvents('databases.[databaseId].collections.[collectionId].indexes.[indexId].delete', [
-            'databaseId' => $database->getId(),
-            'collectionId' => $collection->getId(),
-            'indexId' => $index->getId()
-        ]);
+        $event = 'databases.[databaseId].collections.[collectionId].indexes.[indexId].delete';
        $key = $index->getAttribute('key');
        $status = $index->getAttribute('status', '');
        $project = $dbForPlatform->getDocument('projects', $projectId);
@@ -490,7 +474,7 @@ class Databases extends Action
            throw $e;

        } finally {
-            $this->trigger($database, $collection, $index, $project, $projectId, $events);
+            $this->trigger($database, $collection, $project, $event, $queueForRealtime, null, $index);
            $dbForProject->purgeCachedDocument('database_' . $database->getInternalId(), $collection->getId());
        }
    }
@@ -532,7 +516,6 @@ class Databases extends Action

        $collectionId = $collection->getId();
        $collectionInternalId = $collection->getInternalId();
-        $databaseId = $database->getId();
        $databaseInternalId = $database->getInternalId();

        $dbForProject->deleteCollection('database_' . $databaseInternalId . '_collection_' . $collection->getInternalId());
@@ -568,21 +551,21 @@ class Databases extends Action


    /**
-     * @param string $collection collectionID
+     * @param string $collectionId
     * @param array $queries
     * @param Database $database
     * @param callable|null $callback
     * @return void
     * @throws Exception
     */
-    protected function deleteByGroup(string $collection, array $queries, Database $database, callable $callback = null): void
+    protected function deleteByGroup(string $collectionId, array $queries, Database $database, callable $callback = null): void
    {
        $start = \microtime(true);

        try {
-            $documents = $database->deleteDocuments($collection, $queries);
+            $documents = $database->deleteDocuments($collectionId, $queries);
        } catch (\Throwable $th) {
-            Console::error('Failed to delete documents for collection ' . $collection . ': ' . $th->getMessage());
+            Console::error('Failed to delete documents for collection ' . $collectionId . ': ' . $th->getMessage());
            return;
        }

@@ -598,31 +581,42 @@ class Databases extends Action
        Console::info("Deleted {$count} documents by group in " . ($end - $start) . " seconds");
    }

+    /**
+     * @param Document $database
+     * @param Document $collection
+     * @param Document $project
+     * @param Realtime $queueForRealtime
+     * @param Document|null $attribute
+     * @param Document|null $index
+     * @return void
+     */
    protected function trigger(
        Document $database,
        Document $collection,
-        Document $attribute,
        Document $project,
-        string $projectId,
-        array $events
+        string $event,
+        Realtime $queueForRealtime,
+        Document|null $attribute = null,
+        Document|null $index = null,
    ): void {
-        $target = Realtime::fromPayload(
-            // Pass first, most verbose event pattern
-            event: $events[0],
-            payload: $attribute,
-            project: $project,
-        );
-        Realtime::send(
-            projectId: 'console',
-            payload: $attribute->getArrayCopy(),
-            events: $events,
-            channels: $target['channels'],
-            roles: $target['roles'],
-            options: [
-                'projectId' => $projectId,
-                'databaseId' => $database->getId(),
-                'collectionId' => $collection->getId()
-            ]
-        );
+        $queueForRealtime
+            ->setProject($project)
+            ->setSubscribers(['console'])
+            ->setEvent($event)
+            ->setParam('databaseId', $database->getId())
+            ->setParam('collectionId', $collection->getId());
+
+        if ($attribute !== null && !empty($attribute)) {
+            $queueForRealtime
+                ->setParam('attributeId', $attribute->getId())
+                ->setPayload($attribute->getArrayCopy());
+        }
+        if ($index !== null && !empty($index)) {
+            $queueForRealtime
+                ->setParam('indexId', $index->getId())
+                ->setPayload($index->getArrayCopy());
+        }
+
+        $queueForRealtime->trigger();
    }
 }
@@ -5,8 +5,9 @@ namespace Appwrite\Platform\Workers;
 use Ahc\Jwt\JWT;
 use Appwrite\Event\Event;
 use Appwrite\Event\Func;
+use Appwrite\Event\Realtime;
 use Appwrite\Event\StatsUsage;
-use Appwrite\Messaging\Adapter\Realtime;
+use Appwrite\Event\Webhook;
 use Appwrite\Utopia\Response\Model\Execution;
 use Exception;
 use Executor\Executor;
@@ -44,15 +45,17 @@ class Functions extends Action
            ->inject('project')
            ->inject('message')
            ->inject('dbForProject')
+            ->inject('queueForWebhooks')
            ->inject('queueForFunctions')
+            ->inject('queueForRealtime')
            ->inject('queueForEvents')
            ->inject('queueForStatsUsage')
            ->inject('log')
            ->inject('isResourceBlocked')
-            ->callback(fn (Document $project, Message $message, Database $dbForProject, Func $queueForFunctions, Event $queueForEvents, StatsUsage $queueForStatsUsage, Log $log, callable $isResourceBlocked) => $this->action($project, $message, $dbForProject, $queueForFunctions, $queueForEvents, $queueForStatsUsage, $log, $isResourceBlocked));
+            ->callback(fn (Document $project, Message $message, Database $dbForProject, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Event $queueForEvents, StatsUsage $queueForStatsUsage, Log $log, callable $isResourceBlocked) => $this->action($project, $message, $dbForProject, $queueForWebhooks, $queueForFunctions, $queueForRealtime, $queueForEvents, $queueForStatsUsage, $log, $isResourceBlocked));
    }

-    public function action(Document $project, Message $message, Database $dbForProject, Func $queueForFunctions, Event $queueForEvents, StatsUsage $queueForStatsUsage, Log $log, callable $isResourceBlocked): void
+    public function action(Document $project, Message $message, Database $dbForProject, Webhook $queueForWebhooks, Func $queueForFunctions, Realtime $queueForRealtime, Event $queueForEvents, StatsUsage $queueForStatsUsage, Log $log, callable $isResourceBlocked): void
    {
        $payload = $message->getPayload() ?? [];

@@ -136,7 +139,9 @@ class Functions extends Action
                    $this->execute(
                        log: $log,
                        dbForProject: $dbForProject,
+                        queueForWebhooks: $queueForWebhooks,
                        queueForFunctions: $queueForFunctions,
+                        queueForRealtime: $queueForRealtime,
                        queueForStatsUsage: $queueForStatsUsage,
                        queueForEvents: $queueForEvents,
                        project: $project,
@@ -176,7 +181,9 @@ class Functions extends Action
                $this->execute(
                    log: $log,
                    dbForProject: $dbForProject,
+                    queueForWebhooks: $queueForWebhooks,
                    queueForFunctions: $queueForFunctions,
+                    queueForRealtime: $queueForRealtime,
                    queueForStatsUsage: $queueForStatsUsage,
                    queueForEvents: $queueForEvents,
                    project: $project,
@@ -198,7 +205,9 @@ class Functions extends Action
                $this->execute(
                    log: $log,
                    dbForProject: $dbForProject,
+                    queueForWebhooks: $queueForWebhooks,
                    queueForFunctions: $queueForFunctions,
+                    queueForRealtime: $queueForRealtime,
                    queueForStatsUsage: $queueForStatsUsage,
                    queueForEvents: $queueForEvents,
                    project: $project,
@@ -284,6 +293,7 @@ class Functions extends Action
     * @param Log $log
     * @param Database $dbForProject
     * @param Func $queueForFunctions
+     * @param Realtime $queueForRealtime
     * @param StatsUsage $queueForStatsUsage
     * @param Event $queueForEvents
     * @param Document $project
@@ -307,7 +317,9 @@ class Functions extends Action
    private function execute(
        Log $log,
        Database $dbForProject,
+        Webhook $queueForWebhooks,
        Func $queueForFunctions,
+        Realtime $queueForRealtime,
        StatsUsage $queueForStatsUsage,
        Event $queueForEvents,
        Document $project,
@@ -564,20 +576,20 @@ class Functions extends Action
            ;
        }

-
        $execution = $dbForProject->updateDocument('executions', $executionId, $execution);

-        /** Trigger Webhook */
        $executionModel = new Execution();
        $queueForEvents
-            ->setQueue(Event::WEBHOOK_QUEUE_NAME)
-            ->setClass(Event::WEBHOOK_CLASS_NAME)
            ->setProject($project)
            ->setUser($user)
            ->setEvent('functions.[functionId].executions.[executionId].update')
            ->setParam('functionId', $function->getId())
            ->setParam('executionId', $execution->getId())
-            ->setPayload($execution->getArrayCopy(array_keys($executionModel->getRules())))
+            ->setPayload($execution->getArrayCopy(array_keys($executionModel->getRules())));
+
+        /** Trigger Webhook */
+        $queueForWebhooks
+            ->from($queueForEvents)
            ->trigger();

        /** Trigger Functions */
@@ -585,31 +597,11 @@ class Functions extends Action
            ->from($queueForEvents)
            ->trigger();

-        /** Trigger realtime event */
-        $allEvents = Event::generateEvents('functions.[functionId].executions.[executionId].update', [
-            'functionId' => $function->getId(),
-            'executionId' => $execution->getId()
-        ]);
-        $target = Realtime::fromPayload(
-            // Pass first, most verbose event pattern
-            event: $allEvents[0],
-            payload: $execution,
-            project: $project
-        );
-        Realtime::send(
-            projectId: 'console',
-            payload: $execution->getArrayCopy(),
-            events: $allEvents,
-            channels: $target['channels'],
-            roles: $target['roles']
-        );
-        Realtime::send(
-            projectId: $project->getId(),
-            payload: $execution->getArrayCopy(),
-            events: $allEvents,
-            channels: $target['channels'],
-            roles: $target['roles']
-        );
+        /** Trigger Realtime Events */
+        $queueForRealtime
+            ->from($queueForEvents)
+            ->setSubscribers(['console', $project->getId()])
+            ->trigger();

        if (!empty($error)) {
            throw new Exception($error, $errorCode);
@@ -3,8 +3,7 @@
 namespace Appwrite\Platform\Workers;

 use Ahc\Jwt\JWT;
-use Appwrite\Event\Event;
-use Appwrite\Messaging\Adapter\Realtime;
+use Appwrite\Event\Realtime;
 use Exception;
 use Utopia\CLI\Console;
 use Utopia\Config\Config;
@@ -54,13 +53,14 @@ class Migrations extends Action
            ->inject('dbForProject')
            ->inject('dbForPlatform')
            ->inject('logError')
-            ->callback(fn (Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError) => $this->action($message, $project, $dbForProject, $dbForPlatform, $logError));
+            ->inject('queueForRealtime')
+            ->callback(fn (Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError, Realtime $queueForRealtime) => $this->action($message, $project, $dbForProject, $dbForPlatform, $logError, $queueForRealtime));
    }

    /**
     * @throws Exception
     */
-    public function action(Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError): void
+    public function action(Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError, Realtime $queueForRealtime): void
    {
        $payload = $message->getPayload() ?? [];

@@ -87,7 +87,7 @@ class Migrations extends Action
            return;
        }

-        $this->processMigration($migration);
+        $this->processMigration($migration, $queueForRealtime);
    }

    /**
@@ -155,34 +155,16 @@ class Migrations extends Action
     * @throws \Utopia\Database\Exception
     * @throws Exception
     */
-    protected function updateMigrationDocument(Document $migration, Document $project): Document
+    protected function updateMigrationDocument(Document $migration, Document $project, Realtime $queueForRealtime): Document
    {
-        /** Trigger Realtime */
-        $allEvents = Event::generateEvents('migrations.[migrationId].update', [
-            'migrationId' => $migration->getId(),
-        ]);
-
-        $target = Realtime::fromPayload(
-            event: $allEvents[0],
-            payload: $migration,
-            project: $project
-        );
-
-        Realtime::send(
-            projectId: 'console',
-            payload: $migration->getArrayCopy(),
-            events: $allEvents,
-            channels: $target['channels'],
-            roles: $target['roles'],
-        );
-
-        Realtime::send(
-            projectId: $project->getId(),
-            payload: $migration->getArrayCopy(),
-            events: $allEvents,
-            channels: $target['channels'],
-            roles: $target['roles'],
-        );
+        /** Trigger Realtime Events */
+        $queueForRealtime
+            ->setProject($project)
+            ->setSubscribers(['console', $project->getId()])
+            ->setEvent('migrations.[migrationId].update')
+            ->setParam('migrationId', $migration->getId())
+            ->setPayload($migration->getArrayCopy())
+            ->trigger();

        return $this->dbForProject->updateDocument('migrations', $migration->getId(), $migration);
    }
@@ -231,7 +213,7 @@ class Migrations extends Action
     * @throws \Utopia\Database\Exception
     * @throws Exception
     */
-    protected function processMigration(Document $migration): void
+    protected function processMigration(Document $migration, Realtime $queueForRealtime): void
    {
        $project = $this->project;
        $projectDocument = $this->dbForPlatform->getDocument('projects', $project->getId());
@@ -255,7 +237,7 @@ class Migrations extends Action

            $migration->setAttribute('stage', 'processing');
            $migration->setAttribute('status', 'processing');
-            $this->updateMigrationDocument($migration, $projectDocument);
+            $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime);

            $source = $this->processSource($migration);
            $destination = $this->processDestination($migration, $tempAPIKey);
@@ -269,14 +251,14 @@ class Migrations extends Action

            /** Start Transfer */
            $migration->setAttribute('stage', 'migrating');
-            $this->updateMigrationDocument($migration, $projectDocument);
+            $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime);

            $transfer->run(
                $migration->getAttribute('resources'),
-                function () use ($migration, $transfer, $projectDocument) {
+                function () use ($migration, $transfer, $projectDocument, $queueForRealtime) {
                    $migration->setAttribute('resourceData', json_encode($transfer->getCache()));
                    $migration->setAttribute('statusCounters', json_encode($transfer->getStatusCounters()));
-                    $this->updateMigrationDocument($migration, $projectDocument);
+                    $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime);
                },
                $migration->getAttribute('resourceId'),
                $migration->getAttribute('resourceType')
@@ -313,7 +295,7 @@ class Migrations extends Action
                }

                $migration->setAttribute('errors', $errorMessages);
-                $this->updateMigrationDocument($migration, $projectDocument);
+                $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime);

                return;
            }
@@ -354,7 +336,7 @@ class Migrations extends Action
                $migration->setAttribute('errors', $errorMessages);
            }
        } finally {
-            $this->updateMigrationDocument($migration, $projectDocument);
+            $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime);

            if ($migration->getAttribute('status', '') === 'failed') {
                Console::error('Migration('.$migration->getInternalId().':'.$migration->getId().') failed, Project('.$this->project->getInternalId().':'.$this->project->getId().')');
@@ -117,7 +117,7 @@ class StatsUsageDump extends Action
            $project = new Document($stats['project'] ?? []);

            $numberOfKeys = !empty($stats['keys']) ? count($stats['keys']) : 0;
-            $receivedAt = $stats['receivedAt'] ?? 'NONE';
+            $receivedAt = $stats['receivedAt'] ?? null;
            if ($numberOfKeys === 0) {
                continue;
            }
@@ -134,7 +134,7 @@ class StatsUsageDump extends Action

                    if (str_contains($key, METRIC_DATABASES_STORAGE)) {
                        try {
-                            $this->handleDatabaseStorage($key, $dbForProject, $project);
+                            $this->handleDatabaseStorage($key, $dbForProject, $project, $receivedAt);
                        } catch (\Exception $e) {
                            console::error('[' . DateTime::now() . '] failed to calculate database storage for key [' . $key . '] ' . $e->getMessage());
                        }
@@ -142,7 +142,11 @@ class StatsUsageDump extends Action
                    }

                    foreach ($this->periods as $period => $format) {
-                        $time = 'inf' === $period ? null : date($format, time());
+                        $time = null;
+
+                        if ($period !== 'inf') {
+                            $time = !empty($receivedAt) ? (new \DateTime($receivedAt))->format($format) : date($format, time());
+                        }
                        $id = \md5("{$time}_{$period}_{$key}");

                        $document = new Document([
@@ -171,12 +175,12 @@ class StatsUsageDump extends Action
        }
    }

-    private function handleDatabaseStorage(string $key, Database $dbForProject, Document $project): void
+    private function handleDatabaseStorage(string $key, Database $dbForProject, Document $project, string $receivedAt): void
    {
        $data = explode('.', $key);
        $start = microtime(true);

-        $updateMetric = function (Database $dbForProject, Document $project, int $value, string $key, string $period, string|null $time) {
+        $updateMetric = function (Database $dbForProject, Document $project, int $value, string $key, string $period, string|null $time) use ($receivedAt) {
            $id = \md5("{$time}_{$period}_{$key}");

            $document = new Document([
@@ -197,7 +201,11 @@ class StatsUsageDump extends Action
        };

        foreach ($this->periods as $period => $format) {
-            $time = 'inf' === $period ? null : date($format, time());
+            $time = null;
+
+            if ($period !== 'inf') {
+                $time = !empty($receivedAt) ? (new \DateTime($receivedAt))->format($format) : date($format, time());
+            }
            $id = \md5("{$time}_{$period}_{$key}");

            $value = 0;
@@ -320,7 +328,7 @@ class StatsUsageDump extends Action

    protected function writeToLogsDB(Document $project, Document $document): void
    {
-        if (!System::getEnv('_APP_STATS_USAGE_DUAL_WRITING', false)) {
+        if (System::getEnv('_APP_STATS_USAGE_DUAL_WRITING', 'disabled') === 'disabled') {
            Console::log('Dual Writing is disabled. Skipping...');
            return;
        }
@@ -1,292 +0,0 @@
-<?php
-
-namespace Appwrite\Platform\Workers;
-
-use Appwrite\Event\UsageDump;
-use Exception;
-use Utopia\CLI\Console;
-use Utopia\Database\DateTime;
-use Utopia\Database\Document;
-use Utopia\Platform\Action;
-use Utopia\Queue\Message;
-use Utopia\System\System;
-
-class Usage extends Action
-{
-    private array $stats = [];
-    private int $lastTriggeredTime = 0;
-    private int $aggregationInterval = 20;
-    private int $keys = 0;
-    private const INFINITY_PERIOD = '_inf_';
-    private const KEYS_THRESHOLD = 20000;
-
-    public static function getName(): string
-    {
-        return 'usage';
-    }
-
-    /**
-     * @throws Exception
-     */
-    public function __construct()
-    {
-
-        $this
-        ->desc('Usage worker')
-        ->inject('message')
-        ->inject('project')
-        ->inject('getProjectDB')
-        ->inject('queueForUsageDump')
-        ->callback(function (Message $message, Document $project, callable $getProjectDB, UsageDump $queueForUsageDump) {
-            $this->action($message, $project, $getProjectDB, $queueForUsageDump);
-        });
-
-        $this->aggregationInterval = (int) System::getEnv('_APP_USAGE_AGGREGATION_INTERVAL', '20');
-        $this->lastTriggeredTime = time();
-    }
-
-    /**
-     * @param Message $message
-     * @param Document $project
-     * @param callable $getProjectDB
-     * @param UsageDump $queueForUsageDump
-     * @return void
-     * @throws \Utopia\Database\Exception
-     * @throws Exception
-     */
-    public function action(Message $message, Document $project, callable $getProjectDB, UsageDump $queueForUsageDump): void
-    {
-        $payload = $message->getPayload() ?? [];
-        if (empty($payload)) {
-            throw new Exception('Missing payload');
-        }
-
-
-        if (empty($project->getAttribute('database'))) {
-            var_dump($payload);
-            return;
-        }
-
-        $projectId = $project->getInternalId();
-        foreach ($payload['reduce'] ?? [] as $document) {
-            if (empty($document)) {
-                continue;
-            }
-
-            $this->reduce(
-                project: $project,
-                document: new Document($document),
-                metrics:  $payload['metrics'],
-                getProjectDB: $getProjectDB
-            );
-        }
-
-
-        $this->stats[$projectId]['project'] = [
-            '$id' => $project->getId(),
-            '$internalId' => $project->getInternalId(),
-            'database' => $project->getAttribute('database'),
-        ];
-        $this->stats[$projectId]['receivedAt'] = DateTime::now();
-        foreach ($payload['metrics'] ?? [] as $metric) {
-            $this->keys++;
-            if (!isset($this->stats[$projectId]['keys'][$metric['key']])) {
-                $this->stats[$projectId]['keys'][$metric['key']] = $metric['value'];
-                continue;
-            }
-
-            $this->stats[$projectId]['keys'][$metric['key']] += $metric['value'];
-        }
-
-        // If keys crossed threshold or X time passed since the last send and there are some keys in the array ($this->stats)
-        if (
-            $this->keys >= self::KEYS_THRESHOLD ||
-            (time() - $this->lastTriggeredTime > $this->aggregationInterval  && $this->keys > 0)
-        ) {
-            Console::warning('[' . DateTime::now() . '] Aggregated ' . $this->keys . ' keys');
-
-            $queueForUsageDump
-                ->setStats($this->stats)
-                ->trigger();
-
-            $this->stats = [];
-            $this->keys = 0;
-            $this->lastTriggeredTime = time();
-        }
-    }
-
-    /**
-    * On Documents that tied by relations like functions>deployments>build || documents>collection>database || buckets>files.
-    * When we remove a parent document we need to deduct his children aggregation from the project scope.
-    * @param Document $project
-    * @param Document $document
-    * @param array $metrics
-    * @param  callable $getProjectDB
-    * @return void
-    */
-    private function reduce(Document $project, Document $document, array &$metrics, callable $getProjectDB): void
-    {
-        $dbForProject = $getProjectDB($project);
-
-        try {
-            switch (true) {
-                case $document->getCollection() === 'users': // users
-                    $sessions = count($document->getAttribute(METRIC_SESSIONS, 0));
-                    if (!empty($sessions)) {
-                        $metrics[] = [
-                            'key' => METRIC_SESSIONS,
-                            'value' => ($sessions * -1),
-                        ];
-                    }
-                    break;
-                case $document->getCollection() === 'databases': // databases
-                    $collections = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{databaseInternalId}', $document->getInternalId(), METRIC_DATABASE_ID_COLLECTIONS)));
-                    $documents = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{databaseInternalId}', $document->getInternalId(), METRIC_DATABASE_ID_DOCUMENTS)));
-                    if (!empty($collections['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_COLLECTIONS,
-                            'value' => ($collections['value'] * -1),
-                        ];
-                    }
-
-                    if (!empty($documents['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_DOCUMENTS,
-                            'value' => ($documents['value'] * -1),
-                        ];
-                    }
-                    break;
-                case str_starts_with($document->getCollection(), 'database_') && !str_contains($document->getCollection(), 'collection'): //collections
-                    $parts = explode('_', $document->getCollection());
-                    $databaseInternalId = $parts[1] ?? 0;
-                    $documents = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace(['{databaseInternalId}', '{collectionInternalId}'], [$databaseInternalId, $document->getInternalId()], METRIC_DATABASE_ID_COLLECTION_ID_DOCUMENTS)));
-
-                    if (!empty($documents['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_DOCUMENTS,
-                            'value' => ($documents['value'] * -1),
-                        ];
-                        $metrics[] = [
-                            'key' => str_replace('{databaseInternalId}', $databaseInternalId, METRIC_DATABASE_ID_DOCUMENTS),
-                            'value' => ($documents['value'] * -1),
-                        ];
-                    }
-                    break;
-
-                case $document->getCollection() === 'buckets':
-                    $files = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{bucketInternalId}', $document->getInternalId(), METRIC_BUCKET_ID_FILES)));
-                    $storage = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{bucketInternalId}', $document->getInternalId(), METRIC_BUCKET_ID_FILES_STORAGE)));
-
-                    if (!empty($files['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_FILES,
-                            'value' => ($files['value'] * -1),
-                        ];
-                    }
-
-                    if (!empty($storage['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_FILES_STORAGE,
-                            'value' => ($storage['value'] * -1),
-                        ];
-                    }
-                    break;
-
-                case $document->getCollection() === 'functions':
-                    $deployments = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace(['{resourceType}', '{resourceInternalId}'], ['functions', $document->getInternalId()], METRIC_FUNCTION_ID_DEPLOYMENTS)));
-                    $deploymentsStorage = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace(['{resourceType}', '{resourceInternalId}'], ['functions', $document->getInternalId()], METRIC_FUNCTION_ID_DEPLOYMENTS_STORAGE)));
-                    $builds = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS)));
-                    $buildsSuccess = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_SUCCESS)));
-                    $buildsFailed = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_FAILED)));
-                    $buildsStorage = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_STORAGE)));
-                    $buildsCompute = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_COMPUTE)));
-                    $buildsComputeSuccess = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_COMPUTE_SUCCESS)));
-                    $buildsComputeFailed = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_BUILDS_COMPUTE_FAILED)));
-                    $executions = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_EXECUTIONS)));
-                    $executionsCompute = $dbForProject->getDocument('stats', md5(self::INFINITY_PERIOD . str_replace('{functionInternalId}', $document->getInternalId(), METRIC_FUNCTION_ID_EXECUTIONS_COMPUTE)));
-
-                    if (!empty($deployments['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_DEPLOYMENTS,
-                            'value' => ($deployments['value'] * -1),
-                        ];
-                    }
-
-                    if (!empty($deploymentsStorage['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_DEPLOYMENTS_STORAGE,
-                            'value' => ($deploymentsStorage['value'] * -1),
-                        ];
-                    }
-
-                    if (!empty($builds['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_BUILDS,
-                            'value' => ($builds['value'] * -1),
-                        ];
-                    }
-
-                    if (!empty($buildsSuccess['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_BUILDS_SUCCESS,
-                            'value' => ($buildsSuccess['value'] * -1),
-                        ];
-                    }
-
-                    if (!empty($buildsFailed['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_BUILDS_FAILED,
-                            'value' => ($buildsFailed['value'] * -1),
-                        ];
-                    }
-
-                    if (!empty($buildsStorage['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_BUILDS_STORAGE,
-                            'value' => ($buildsStorage['value'] * -1),
-                        ];
-                    }
-
-                    if (!empty($buildsCompute['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_BUILDS_COMPUTE,
-                            'value' => ($buildsCompute['value'] * -1),
-                        ];
-                    }
-
-                    if (!empty($buildsComputeSuccess['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_BUILDS_COMPUTE_SUCCESS,
-                            'value' => ($buildsComputeSuccess['value'] * -1),
-                        ];
-                    }
-
-                    if (!empty($buildsComputeFailed['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_BUILDS_COMPUTE_FAILED,
-                            'value' => ($buildsComputeFailed['value'] * -1),
-                        ];
-                    }
-
-                    if (!empty($executions['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_EXECUTIONS,
-                            'value' => ($executions['value'] * -1),
-                        ];
-                    }
-
-                    if (!empty($executionsCompute['value'])) {
-                        $metrics[] = [
-                            'key' => METRIC_EXECUTIONS_COMPUTE,
-                            'value' => ($executionsCompute['value'] * -1),
-                        ];
-                    }
-                    break;
-                default:
-                    break;
-            }
-        } catch (\Throwable $e) {
-            console::error("[reducer] " . " {DateTime::now()} " . " {$project->getInternalId()} " . " {$e->getMessage()}");
-        }
-    }
-}
@@ -1,286 +0,0 @@
-<?php
-
-namespace Appwrite\Platform\Workers;
-
-use Appwrite\Extend\Exception;
-use Utopia\CLI\Console;
-use Utopia\Database\Database;
-use Utopia\Database\DateTime;
-use Utopia\Database\Document;
-use Utopia\Database\Exception\Duplicate;
-use Utopia\Platform\Action;
-use Utopia\Queue\Message;
-use Utopia\System\System;
-
-const METRIC_COLLECTION_LEVEL_STORAGE = 4;
-const METRIC_DATABASE_LEVEL_STORAGE = 3;
-const METRIC_PROJECT_LEVEL_STORAGE = 2;
-
-class UsageDump extends Action
-{
-    protected array $stats = [];
-    protected array $periods = [
-        '1h' => 'Y-m-d H:00',
-        '1d' => 'Y-m-d 00:00',
-        'inf' => '0000-00-00 00:00'
-    ];
-
-    public static function getName(): string
-    {
-        return 'usage-dump';
-    }
-
-    /**
-     * @throws \Exception
-     */
-    public function __construct()
-    {
-        $this
-            ->inject('message')
-            ->inject('getProjectDB')
-            ->callback(function (Message $message, callable $getProjectDB) {
-                $this->action($message, $getProjectDB);
-            });
-    }
-
-    /**
-     * @param Message $message
-     * @param callable $getProjectDB
-     * @return void
-     * @throws Exception
-     * @throws \Utopia\Database\Exception
-     */
-    public function action(Message $message, callable $getProjectDB): void
-    {
-        $payload = $message->getPayload() ?? [];
-        if (empty($payload)) {
-            throw new Exception('Missing payload');
-        }
-
-
-        foreach ($payload['stats'] ?? [] as $stats) {
-            $project = new Document($stats['project'] ?? []);
-
-            /**
-             * End temp bug fallback
-             */
-            $numberOfKeys = !empty($stats['keys']) ? count($stats['keys']) : 0;
-            $receivedAt = $stats['receivedAt'] ?? 'NONE';
-            if ($numberOfKeys === 0) {
-                continue;
-            }
-
-            console::log('['.DateTime::now().'] Id: '.$project->getId(). ' InternalId: '.$project->getInternalId(). ' Db: '.$project->getAttribute('database').' ReceivedAt: '.$receivedAt. ' Keys: '.$numberOfKeys);
-
-            try {
-                $dbForProject = $getProjectDB($project);
-                foreach ($stats['keys'] ?? [] as $key => $value) {
-                    if ($value == 0) {
-                        continue;
-                    }
-
-                    if (str_contains($key, METRIC_DATABASES_STORAGE)) {
-                        try {
-                            $this->handleDatabaseStorage($key, $dbForProject);
-                        } catch (\Exception $e) {
-                            console::error('[' . DateTime::now() . '] failed to calculate database storage for key [' . $key . '] ' . $e->getMessage());
-                        }
-                        continue;
-                    }
-
-                    foreach ($this->periods as $period => $format) {
-                        $time = 'inf' === $period ? null : date($format, time());
-                        $id = \md5("{$time}_{$period}_{$key}");
-
-                        try {
-                            $dbForProject->createDocument('stats', new Document([
-                                '$id' => $id,
-                                'period' => $period,
-                                'time' => $time,
-                                'metric' => $key,
-                                'value' => $value,
-                                'region' => System::getEnv('_APP_REGION', 'default'),
-                            ]));
-                        } catch (Duplicate $th) {
-                            if ($value < 0) {
-                                $dbForProject->decreaseDocumentAttribute(
-                                    'stats',
-                                    $id,
-                                    'value',
-                                    abs($value)
-                                );
-                            } else {
-                                $dbForProject->increaseDocumentAttribute(
-                                    'stats',
-                                    $id,
-                                    'value',
-                                    $value
-                                );
-                            }
-                        }
-                    }
-                }
-            } catch (\Exception $e) {
-                console::error('[' . DateTime::now() . '] project [' . $project->getInternalId() . '] database [' . $project['database'] . '] ' . ' ' . $e->getMessage());
-            }
-        }
-    }
-
-    private function handleDatabaseStorage(string $key, Database $dbForProject): void
-    {
-        $data = explode('.', $key);
-        $start = microtime(true);
-
-        $updateMetric = function (Database $dbForProject, int $value, string $key, string $period, string|null $time) {
-            $id = \md5("{$time}_{$period}_{$key}");
-
-            try {
-                $dbForProject->createDocument('stats', new Document([
-                    '$id' => $id,
-                    'period' => $period,
-                    'time' => $time,
-                    'metric' => $key,
-                    'value' => $value,
-                    'region' => System::getEnv('_APP_REGION', 'default'),
-                ]));
-            } catch (Duplicate $th) {
-                if ($value < 0) {
-                    $dbForProject->decreaseDocumentAttribute(
-                        'stats',
-                        $id,
-                        'value',
-                        abs($value)
-                    );
-                } else {
-                    $dbForProject->increaseDocumentAttribute(
-                        'stats',
-                        $id,
-                        'value',
-                        $value
-                    );
-                }
-            }
-        };
-
-        foreach ($this->periods as $period => $format) {
-            $time = 'inf' === $period ? null : date($format, time());
-            $id = \md5("{$time}_{$period}_{$key}");
-
-            $value = 0;
-            $previousValue = 0;
-            try {
-                $previousValue = ($dbForProject->getDocument('stats', $id))->getAttribute('value', 0);
-            } catch (\Exception $e) {
-                // No previous value
-            }
-
-            switch (count($data)) {
-                // Collection Level
-                case METRIC_COLLECTION_LEVEL_STORAGE:
-                    Console::log('[' . DateTime::now() . '] Collection Level Storage Calculation [' . $key . ']');
-                    $databaseInternalId = $data[0];
-                    $collectionInternalId = $data[1];
-
-                    try {
-                        $value = $dbForProject->getSizeOfCollection('database_' . $databaseInternalId . '_collection_' . $collectionInternalId);
-                    } catch (\Exception $e) {
-                        // Collection not found
-                        if ($e->getMessage() !== 'Collection not found') {
-                            throw $e;
-                        }
-                    }
-
-                    // Compare with previous value
-                    $diff = $value - $previousValue;
-
-                    if ($diff === 0) {
-                        break;
-                    }
-
-                    // Update Collection
-                    $updateMetric($dbForProject, $diff, $key, $period, $time);
-
-                    // Update Database
-                    $databaseKey = str_replace(['{databaseInternalId}'], [$data[0]], METRIC_DATABASE_ID_STORAGE);
-                    $updateMetric($dbForProject, $diff, $databaseKey, $period, $time);
-
-                    // Update Project
-                    $projectKey = METRIC_DATABASES_STORAGE;
-                    $updateMetric($dbForProject, $diff, $projectKey, $period, $time);
-                    break;
-                    // Database Level
-                case METRIC_DATABASE_LEVEL_STORAGE:
-                    Console::log('[' . DateTime::now() . '] Database Level Storage Calculation [' . $key . ']');
-                    $databaseInternalId = $data[0];
-
-                    $collections = [];
-                    try {
-                        $collections = $dbForProject->find('database_' . $databaseInternalId);
-                    } catch (\Exception $e) {
-                        // Database not found
-                        if ($e->getMessage() !== 'Collection not found') {
-                            throw $e;
-                        }
-                    }
-
-                    foreach ($collections as $collection) {
-                        try {
-                            $value += $dbForProject->getSizeOfCollection('database_' . $databaseInternalId . '_collection_' . $collection->getInternalId());
-                        } catch (\Exception $e) {
-                            // Collection not found
-                            if ($e->getMessage() !== 'Collection not found') {
-                                throw $e;
-                            }
-                        }
-                    }
-
-                    $diff = $value - $previousValue;
-
-                    if ($diff === 0) {
-                        break;
-                    }
-
-                    // Update Database
-                    $databaseKey = str_replace(['{databaseInternalId}'], [$data[0]], METRIC_DATABASE_ID_STORAGE);
-                    $updateMetric($dbForProject, $diff, $databaseKey, $period, $time);
-
-                    // Update Project
-                    $projectKey = METRIC_DATABASES_STORAGE;
-                    $updateMetric($dbForProject, $diff, $projectKey, $period, $time);
-                    break;
-                    // Project Level
-                case METRIC_PROJECT_LEVEL_STORAGE:
-                    Console::log('[' . DateTime::now() . '] Project Level Storage Calculation [' . $key . ']');
-                    // Get all project databases
-                    $databases = $dbForProject->find('database');
-
-                    // Recalculate all databases
-                    foreach ($databases as $database) {
-                        $collections = $dbForProject->find('database_' . $database->getInternalId());
-
-                        foreach ($collections as $collection) {
-                            try {
-                                $value += $dbForProject->getSizeOfCollection('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId());
-                            } catch (\Exception $e) {
-                                // Collection not found
-                                if ($e->getMessage() !== 'Collection not found') {
-                                    throw $e;
-                                }
-                            }
-                        }
-                    }
-
-                    $diff = $value - $previousValue;
-
-                    // Update Project
-                    $projectKey = METRIC_DATABASES_STORAGE;
-                    $updateMetric($dbForProject, $diff, $projectKey, $period, $time);
-                    break;
-            }
-        }
-
-        $end = microtime(true);
-
-        console::log('[' . DateTime::now() . '] DB Storage Calculation [' . $key . '] took ' . (($end - $start) * 1000) . ' milliseconds');
-    }
-}