diff --git a/app/cli.php b/app/cli.php
index 47f4525f0b..abf75c9608 100644
--- a/app/cli.php
+++ b/app/cli.php
@@ -19,7 +19,7 @@ use Utopia\DSN\DSN;
 use Utopia\Logger\Log;
 use Utopia\Platform\Service;
 use Utopia\Pools\Group;
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
 use Utopia\Registry\Registry;
 use Utopia\System\System;
 
@@ -160,18 +160,18 @@ CLI::setResource('getProjectDB', function (Group $pools, Database $dbForPlatform
     };
 }, ['pools', 'dbForPlatform', 'cache']);
 
-CLI::setResource('queue', function (Group $pools) {
-    return $pools->get('queue')->pop()->getResource();
+CLI::setResource('publisher', function (Group $pools) {
+    return $pools->get('publisher')->pop()->getResource();
 }, ['pools']);
-CLI::setResource('queueForFunctions', function (Connection $queue) {
-    return new Func($queue);
-}, ['queue']);
-CLI::setResource('queueForDeletes', function (Connection $queue) {
-    return new Delete($queue);
-}, ['queue']);
-CLI::setResource('queueForCertificates', function (Connection $queue) {
-    return new Certificate($queue);
-}, ['queue']);
+CLI::setResource('queueForFunctions', function (Publisher $publisher) {
+    return new Func($publisher);
+}, ['publisher']);
+CLI::setResource('queueForDeletes', function (Publisher $publisher) {
+    return new Delete($publisher);
+}, ['publisher']);
+CLI::setResource('queueForCertificates', function (Publisher $publisher) {
+    return new Certificate($publisher);
+}, ['publisher']);
 CLI::setResource('logError', function (Registry $register) {
     return function (Throwable $error, string $namespace, string $action) use ($register) {
         $logger = $register->get('logger');
diff --git a/app/controllers/api/health.php b/app/controllers/api/health.php
index 1db4713311..305444abc5 100644
--- a/app/controllers/api/health.php
+++ b/app/controllers/api/health.php
@@ -13,8 +13,8 @@ use Utopia\Config\Config;
 use Utopia\Database\Document;
 use Utopia\Domains\Validator\PublicDomain;
 use Utopia\Pools\Group;
-use Utopia\Queue\Client;
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
+use Utopia\Queue\Queue;
 use Utopia\Registry\Registry;
 use Utopia\Storage\Device;
 use Utopia\Storage\Device\Local;
@@ -188,69 +188,6 @@ App::get('/v1/health/cache')
         ]), Response::MODEL_HEALTH_STATUS_LIST);
     });
 
-App::get('/v1/health/queue')
-    ->desc('Get queue')
-    ->groups(['api', 'health'])
-    ->label('scope', 'health.read')
-    ->label('sdk', new Method(
-        auth: [AuthType::KEY],
-        namespace: 'health',
-        name: 'getQueue',
-        description: '/docs/references/health/get-queue.md',
-        responses: [
-            new SDKResponse(
-                code: Response::STATUS_CODE_OK,
-                model: Response::MODEL_HEALTH_STATUS,
-            )
-        ],
-        contentType: ContentType::JSON
-    ))
-    ->inject('response')
-    ->inject('pools')
-    ->action(function (Response $response, Group $pools) {
-
-        $output = [];
-
-        $configs = [
-            'Queue' => Config::getParam('pools-queue'),
-        ];
-
-        foreach ($configs as $key => $config) {
-            foreach ($config as $database) {
-                $checkStart = \microtime(true);
-                try {
-                    /** @var Connection $adapter */
-                    $adapter = $pools->get($database)->pop()->getResource();
-
-                    if ($adapter->ping()) {
-                        $output[] = new Document([
-                            'name' => $key . " ($database)",
-                            'status' => 'pass',
-                            'ping' => \round((\microtime(true) - $checkStart) / 1000)
-                        ]);
-                    } else {
-                        $output[] = new Document([
-                            'name' => $key . " ($database)",
-                            'status' => 'fail',
-                            'ping' => \round((\microtime(true) - $checkStart) / 1000)
-                        ]);
-                    }
-                } catch (\Throwable $th) {
-                    $output[] = new Document([
-                        'name' => $key . " ($database)",
-                        'status' => 'fail',
-                        'ping' => \round((\microtime(true) - $checkStart) / 1000)
-                    ]);
-                }
-            }
-        }
-
-        $response->dynamic(new Document([
-            'statuses' => $output,
-            'total' => count($output),
-        ]), Response::MODEL_HEALTH_STATUS_LIST);
-    });
-
 App::get('/v1/health/pubsub')
     ->desc('Get pubsub')
     ->groups(['api', 'health'])
@@ -396,13 +333,12 @@ App::get('/v1/health/queue/webhooks')
         contentType: ContentType::JSON
     ))
     ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true)
-    ->inject('queue')
+    ->inject('publisher')
     ->inject('response')
-    ->action(function (int|string $threshold, Connection $queue, Response $response) {
+    ->action(function (int|string $threshold, Publisher $publisher, Response $response) {
         $threshold = \intval($threshold);
 
-        $client = new Client(Event::WEBHOOK_QUEUE_NAME, $queue);
-        $size = $client->getQueueSize();
+        $size = $publisher->getQueueSize(new Queue(Event::WEBHOOK_QUEUE_NAME));
 
         if ($size >= $threshold) {
             throw new Exception(Exception::HEALTH_QUEUE_SIZE_EXCEEDED, "Queue size threshold hit. Current size is {$size} and threshold is {$threshold}.");
@@ -429,13 +365,12 @@ App::get('/v1/health/queue/logs')
         contentType: ContentType::JSON
     ))
     ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true)
-    ->inject('queue')
+    ->inject('publisher')
     ->inject('response')
-    ->action(function (int|string $threshold, Connection $queue, Response $response) {
+    ->action(function (int|string $threshold, Publisher $publisher, Response $response) {
         $threshold = \intval($threshold);
 
-        $client = new Client(Event::AUDITS_QUEUE_NAME, $queue);
-        $size = $client->getQueueSize();
+        $size = $publisher->getQueueSize(new Queue(Event::AUDITS_QUEUE_NAME));
 
         if ($size >= $threshold) {
             throw new Exception(Exception::HEALTH_QUEUE_SIZE_EXCEEDED, "Queue size threshold hit. Current size is {$size} and threshold is {$threshold}.");
@@ -518,13 +453,12 @@ App::get('/v1/health/queue/certificates')
         contentType: ContentType::JSON
     ))
     ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true)
-    ->inject('queue')
+    ->inject('publisher')
     ->inject('response')
-    ->action(function (int|string $threshold, Connection $queue, Response $response) {
+    ->action(function (int|string $threshold, Publisher $publisher, Response $response) {
         $threshold = \intval($threshold);
 
-        $client = new Client(Event::CERTIFICATES_QUEUE_NAME, $queue);
-        $size = $client->getQueueSize();
+        $size = $publisher->getQueueSize(new Queue(Event::CERTIFICATES_QUEUE_NAME));
 
         if ($size >= $threshold) {
             throw new Exception(Exception::HEALTH_QUEUE_SIZE_EXCEEDED, "Queue size threshold hit. Current size is {$size} and threshold is {$threshold}.");
@@ -551,13 +485,12 @@ App::get('/v1/health/queue/builds')
         contentType: ContentType::JSON
     ))
     ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true)
-    ->inject('queue')
+    ->inject('publisher')
     ->inject('response')
-    ->action(function (int|string $threshold, Connection $queue, Response $response) {
+    ->action(function (int|string $threshold, Publisher $publisher, Response $response) {
         $threshold = \intval($threshold);
 
-        $client = new Client(Event::BUILDS_QUEUE_NAME, $queue);
-        $size = $client->getQueueSize();
+        $size = $publisher->getQueueSize(new Queue(Event::BUILDS_QUEUE_NAME));
 
         if ($size >= $threshold) {
             throw new Exception(Exception::HEALTH_QUEUE_SIZE_EXCEEDED, "Queue size threshold hit. Current size is {$size} and threshold is {$threshold}.");
@@ -585,13 +518,12 @@ App::get('/v1/health/queue/databases')
     ))
     ->param('name', 'database_db_main', new Text(256), 'Queue name for which to check the queue size', true)
     ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true)
-    ->inject('queue')
+    ->inject('publisher')
     ->inject('response')
-    ->action(function (string $name, int|string $threshold, Connection $queue, Response $response) {
+    ->action(function (string $name, int|string $threshold, Publisher $publisher, Response $response) {
         $threshold = \intval($threshold);
 
-        $client = new Client($name, $queue);
-        $size = $client->getQueueSize();
+        $size = $publisher->getQueueSize(new Queue($name));
 
         if ($size >= $threshold) {
             throw new Exception(Exception::HEALTH_QUEUE_SIZE_EXCEEDED, "Queue size threshold hit. Current size is {$size} and threshold is {$threshold}.");
@@ -618,13 +550,12 @@ App::get('/v1/health/queue/deletes')
         contentType: ContentType::JSON
     ))
     ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true)
-    ->inject('queue')
+    ->inject('publisher')
     ->inject('response')
-    ->action(function (int|string $threshold, Connection $queue, Response $response) {
+    ->action(function (int|string $threshold, Publisher $publisher, Response $response) {
         $threshold = \intval($threshold);
 
-        $client = new Client(Event::DELETE_QUEUE_NAME, $queue);
-        $size = $client->getQueueSize();
+        $size = $publisher->getQueueSize(new Queue(Event::DELETE_QUEUE_NAME));
 
         if ($size >= $threshold) {
             throw new Exception(Exception::HEALTH_QUEUE_SIZE_EXCEEDED, "Queue size threshold hit. Current size is {$size} and threshold is {$threshold}.");
@@ -651,13 +582,12 @@ App::get('/v1/health/queue/mails')
         contentType: ContentType::JSON
     ))
     ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true)
-    ->inject('queue')
+    ->inject('publisher')
     ->inject('response')
-    ->action(function (int|string $threshold, Connection $queue, Response $response) {
+    ->action(function (int|string $threshold, Publisher $publisher, Response $response) {
         $threshold = \intval($threshold);
 
-        $client = new Client(Event::MAILS_QUEUE_NAME, $queue);
-        $size = $client->getQueueSize();
+        $size = $publisher->getQueueSize(new Queue(Event::MAILS_QUEUE_NAME));
 
         if ($size >= $threshold) {
             throw new Exception(Exception::HEALTH_QUEUE_SIZE_EXCEEDED, "Queue size threshold hit. Current size is {$size} and threshold is {$threshold}.");
@@ -684,13 +614,12 @@ App::get('/v1/health/queue/messaging')
         contentType: ContentType::JSON
     ))
     ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true)
-    ->inject('queue')
+    ->inject('publisher')
     ->inject('response')
-    ->action(function (int|string $threshold, Connection $queue, Response $response) {
+    ->action(function (int|string $threshold, Publisher $publisher, Response $response) {
         $threshold = \intval($threshold);
 
-        $client = new Client(Event::MESSAGING_QUEUE_NAME, $queue);
-        $size = $client->getQueueSize();
+        $size = $publisher->getQueueSize(new Queue(Event::MESSAGING_QUEUE_NAME));
 
         if ($size >= $threshold) {
             throw new Exception(Exception::HEALTH_QUEUE_SIZE_EXCEEDED, "Queue size threshold hit. Current size is {$size} and threshold is {$threshold}.");
@@ -717,13 +646,12 @@ App::get('/v1/health/queue/migrations')
         contentType: ContentType::JSON
     ))
     ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true)
-    ->inject('queue')
+    ->inject('publisher')
     ->inject('response')
-    ->action(function (int|string $threshold, Connection $queue, Response $response) {
+    ->action(function (int|string $threshold, Publisher $publisher, Response $response) {
         $threshold = \intval($threshold);
 
-        $client = new Client(Event::MIGRATIONS_QUEUE_NAME, $queue);
-        $size = $client->getQueueSize();
+        $size = $publisher->getQueueSize(new Queue(Event::MIGRATIONS_QUEUE_NAME));
 
         if ($size >= $threshold) {
             throw new Exception(Exception::HEALTH_QUEUE_SIZE_EXCEEDED, "Queue size threshold hit. Current size is {$size} and threshold is {$threshold}.");
@@ -750,13 +678,12 @@ App::get('/v1/health/queue/functions')
         contentType: ContentType::JSON
     ))
     ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true)
-    ->inject('queue')
+    ->inject('publisher')
     ->inject('response')
-    ->action(function (int|string $threshold, Connection $queue, Response $response) {
+    ->action(function (int|string $threshold, Publisher $publisher, Response $response) {
         $threshold = \intval($threshold);
 
-        $client = new Client(Event::FUNCTIONS_QUEUE_NAME, $queue);
-        $size = $client->getQueueSize();
+        $size = $publisher->getQueueSize(new Queue(Event::FUNCTIONS_QUEUE_NAME));
 
         if ($size >= $threshold) {
             throw new Exception(Exception::HEALTH_QUEUE_SIZE_EXCEEDED, "Queue size threshold hit. Current size is {$size} and threshold is {$threshold}.");
@@ -783,13 +710,12 @@ App::get('/v1/health/queue/usage')
         contentType: ContentType::JSON
     ))
     ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true)
-    ->inject('queue')
+    ->inject('publisher')
     ->inject('response')
-    ->action(function (int|string $threshold, Connection $queue, Response $response) {
+    ->action(function (int|string $threshold, Publisher $publisher, Response $response) {
         $threshold = \intval($threshold);
 
-        $client = new Client(Event::USAGE_QUEUE_NAME, $queue);
-        $size = $client->getQueueSize();
+        $size = $publisher->getQueueSize(new Queue(Event::USAGE_QUEUE_NAME));
 
         if ($size >= $threshold) {
             throw new Exception(Exception::HEALTH_QUEUE_SIZE_EXCEEDED, "Queue size threshold hit. Current size is {$size} and threshold is {$threshold}.");
@@ -816,13 +742,12 @@ App::get('/v1/health/queue/usage-dump')
         contentType: ContentType::JSON
     ))
     ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true)
-    ->inject('queue')
+    ->inject('publisher')
     ->inject('response')
-    ->action(function (int|string $threshold, Connection $queue, Response $response) {
+    ->action(function (int|string $threshold, Publisher $publisher, Response $response) {
         $threshold = \intval($threshold);
 
-        $client = new Client(Event::USAGE_DUMP_QUEUE_NAME, $queue);
-        $size = $client->getQueueSize();
+        $size = $publisher->getQueueSize(new Queue(Event::USAGE_DUMP_QUEUE_NAME));
 
         if ($size >= $threshold) {
             throw new Exception(Exception::HEALTH_QUEUE_SIZE_EXCEEDED, "Queue size threshold hit. Current size is {$size} and threshold is {$threshold}.");
@@ -1005,12 +930,11 @@ App::get('/v1/health/queue/failed/:name')
     ]), 'The name of the queue')
     ->param('threshold', 5000, new Integer(true), 'Queue size threshold. When hit (equal or higher), endpoint returns server error. Default value is 5000.', true)
     ->inject('response')
-    ->inject('queue')
-    ->action(function (string $name, int|string $threshold, Response $response, Connection $queue) {
+    ->inject('publisher')
+    ->action(function (string $name, int|string $threshold, Response $response, Publisher $publisher) {
         $threshold = \intval($threshold);
 
-        $client = new Client($name, $queue);
-        $failed = $client->countFailedJobs();
+        $failed = $publisher->getQueueSize(new Queue($name), failedJobs: true);
 
         if ($failed >= $threshold) {
             throw new Exception(Exception::HEALTH_QUEUE_SIZE_EXCEEDED, "Queue failed jobs threshold hit. Current size is {$failed} and threshold is {$threshold}.");
diff --git a/app/controllers/shared/api.php b/app/controllers/shared/api.php
index d8f1eb59ef..bce36e9303 100644
--- a/app/controllers/shared/api.php
+++ b/app/controllers/shared/api.php
@@ -28,7 +28,7 @@ use Utopia\Database\DateTime;
 use Utopia\Database\Document;
 use Utopia\Database\Helpers\Role;
 use Utopia\Database\Validator\Authorization;
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
 use Utopia\System\System;
 use Utopia\Validator\WhiteList;
 
@@ -429,7 +429,7 @@ App::init()
     ->inject('response')
     ->inject('project')
     ->inject('user')
-    ->inject('queue')
+    ->inject('publisher')
     ->inject('queueForEvents')
     ->inject('queueForMessaging')
     ->inject('queueForAudits')
@@ -440,7 +440,7 @@ App::init()
     ->inject('dbForProject')
     ->inject('timelimit')
     ->inject('mode')
-    ->action(function (App $utopia, Request $request, Response $response, Document $project, Document $user, Connection $queue, Event $queueForEvents, Messaging $queueForMessaging, Audit $queueForAudits, Delete $queueForDeletes, EventDatabase $queueForDatabase, Build $queueForBuilds, Usage $queueForUsage, Database $dbForProject, callable $timelimit, string $mode) use ($usageDatabaseListener, $eventDatabaseListener) {
+    ->action(function (App $utopia, Request $request, Response $response, Document $project, Document $user, Publisher $publisher, Event $queueForEvents, Messaging $queueForMessaging, Audit $queueForAudits, Delete $queueForDeletes, EventDatabase $queueForDatabase, Build $queueForBuilds, Usage $queueForUsage, Database $dbForProject, callable $timelimit, string $mode) use ($usageDatabaseListener, $eventDatabaseListener) {
 
         $route = $utopia->getRoute();
 
@@ -544,9 +544,9 @@ App::init()
 
         // Clone the queues, to prevent events triggered by the database listener
         // from overwriting the events that are supposed to be triggered in the shutdown hook.
-        $queueForEventsClone = new Event($queue);
-        $queueForFunctions = new Func($queue);
-        $queueForWebhooks = new Webhook($queue);
+        $queueForEventsClone = new Event($publisher);
+        $queueForFunctions = new Func($publisher);
+        $queueForWebhooks = new Webhook($publisher);
         $queueForRealtime = new Realtime();
 
         $dbForProject
diff --git a/app/init.php b/app/init.php
index e912ffd8f4..941e253260 100644
--- a/app/init.php
+++ b/app/init.php
@@ -77,7 +77,6 @@ use Utopia\Logger\Logger;
 use Utopia\Pools\Group;
 use Utopia\Pools\Pool;
 use Utopia\Queue;
-use Utopia\Queue\Connection;
 use Utopia\Registry\Registry;
 use Utopia\Storage\Device;
 use Utopia\Storage\Device\Backblaze;
@@ -890,8 +889,14 @@ $register->set('pools', function () {
             'multiple' => false,
             'schemes' => ['mariadb', 'mysql'],
         ],
-        'queue' => [
-            'type' => 'queue',
+        'publisher' => [
+            'type' => 'publisher',
+            'dsns' => $fallbackForRedis,
+            'multiple' => false,
+            'schemes' => ['redis'],
+        ],
+        'consumer' => [
+            'type' => 'consumer',
             'dsns' => $fallbackForRedis,
             'multiple' => false,
             'schemes' => ['redis'],
@@ -999,31 +1004,26 @@ $register->set('pools', function () {
                         };
 
                         $adapter->setDatabase($dsn->getPath());
-                        break;
+                        return $adapter;
                     case 'pubsub':
-                        $adapter = match ($dsn->getScheme()) {
+                        return match ($dsn->getScheme()) {
                             'redis' => new PubSub($resource()),
                             default => null
                         };
-                        break;
-                    case 'queue':
-                        $adapter = match ($dsn->getScheme()) {
-                            'redis' => new Queue\Connection\Redis($dsn->getHost(), $dsn->getPort()),
+                    case 'publisher':
+                    case 'consumer':
+                        return match ($dsn->getScheme()) {
+                            'redis' => new Queue\Broker\Redis(new Queue\Connection\Redis($dsn->getHost(), $dsn->getPort())),
                             default => null
                         };
-                        break;
                     case 'cache':
-                        $adapter = match ($dsn->getScheme()) {
+                        return match ($dsn->getScheme()) {
                             'redis' => new RedisCache($resource()),
                             default => null
                         };
-                        break;
-
                     default:
                         throw new Exception(Exception::GENERAL_SERVER_ERROR, "Server error: Missing adapter implementation.");
                 }
-
-                return $adapter;
             });
 
             $group->add($pool);
@@ -1146,48 +1146,51 @@ App::setResource('localeCodes', function () {
 });
 
 // Queues
-App::setResource('queue', function (Group $pools) {
-    return $pools->get('queue')->pop()->getResource();
+App::setResource('publisher', function (Group $pools) {
+    return $pools->get('publisher')->pop()->getResource();
 }, ['pools']);
-App::setResource('queueForMessaging', function (Connection $queue) {
-    return new Messaging($queue);
-}, ['queue']);
-App::setResource('queueForMails', function (Connection $queue) {
-    return new Mail($queue);
-}, ['queue']);
-App::setResource('queueForBuilds', function (Connection $queue) {
-    return new Build($queue);
-}, ['queue']);
-App::setResource('queueForDatabase', function (Connection $queue) {
-    return new EventDatabase($queue);
-}, ['queue']);
-App::setResource('queueForDeletes', function (Connection $queue) {
-    return new Delete($queue);
-}, ['queue']);
-App::setResource('queueForEvents', function (Connection $queue) {
-    return new Event($queue);
-}, ['queue']);
-App::setResource('queueForWebhooks', function (Connection $queue) {
-    return new Webhook($queue);
-}, ['queue']);
+App::setResource('consumer', function (Group $pools) {
+    return $pools->get('consumer')->pop()->getResource();
+}, ['pools']);
+App::setResource('queueForMessaging', function (Queue\Publisher $publisher) {
+    return new Messaging($publisher);
+}, ['publisher']);
+App::setResource('queueForMails', function (Queue\Publisher $publisher) {
+    return new Mail($publisher);
+}, ['publisher']);
+App::setResource('queueForBuilds', function (Queue\Publisher $publisher) {
+    return new Build($publisher);
+}, ['publisher']);
+App::setResource('queueForDatabase', function (Queue\Publisher $publisher) {
+    return new EventDatabase($publisher);
+}, ['publisher']);
+App::setResource('queueForDeletes', function (Queue\Publisher $publisher) {
+    return new Delete($publisher);
+}, ['publisher']);
+App::setResource('queueForEvents', function (Queue\Publisher $publisher) {
+    return new Event($publisher);
+}, ['publisher']);
+App::setResource('queueForWebhooks', function (Queue\Publisher $publisher) {
+    return new Webhook($publisher);
+}, ['publisher']);
 App::setResource('queueForRealtime', function () {
     return new Realtime();
 }, []);
-App::setResource('queueForAudits', function (Connection $queue) {
-    return new Audit($queue);
-}, ['queue']);
-App::setResource('queueForFunctions', function (Connection $queue) {
-    return new Func($queue);
-}, ['queue']);
-App::setResource('queueForUsage', function (Connection $queue) {
-    return new Usage($queue);
-}, ['queue']);
-App::setResource('queueForCertificates', function (Connection $queue) {
-    return new Certificate($queue);
-}, ['queue']);
-App::setResource('queueForMigrations', function (Connection $queue) {
-    return new Migration($queue);
-}, ['queue']);
+App::setResource('queueForAudits', function (Queue\Publisher $publisher) {
+    return new Audit($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']);
+App::setResource('queueForMigrations', function (Queue\Publisher $publisher) {
+    return new Migration($publisher);
+}, ['publisher']);
 App::setResource('clients', function ($request, $console, $project) {
     $console->setAttribute('platforms', [ // Always allow current host
         '$collection' => ID::custom('platforms'),
diff --git a/app/worker.php b/app/worker.php
index 6eb1363e9b..854d9cb8ae 100644
--- a/app/worker.php
+++ b/app/worker.php
@@ -31,8 +31,8 @@ use Utopia\Logger\Log;
 use Utopia\Logger\Logger;
 use Utopia\Platform\Service;
 use Utopia\Pools\Group;
-use Utopia\Queue\Connection;
 use Utopia\Queue\Message;
+use Utopia\Queue\Publisher;
 use Utopia\Queue\Server;
 use Utopia\Registry\Registry;
 use Utopia\System\System;
@@ -224,57 +224,61 @@ Server::setResource('timelimit', function (\Redis $redis) {
 
 Server::setResource('log', fn () => new Log());
 
-Server::setResource('queueForUsage', function (Connection $queue) {
-    return new Usage($queue);
-}, ['queue']);
-
-Server::setResource('queueForUsageDump', function (Connection $queue) {
-    return new UsageDump($queue);
-}, ['queue']);
-
-Server::setResource('queue', function (Group $pools) {
-    return $pools->get('queue')->pop()->getResource();
+Server::setResource('publisher', function (Group $pools) {
+    return $pools->get('publisher')->pop()->getResource();
 }, ['pools']);
 
-Server::setResource('queueForDatabase', function (Connection $queue) {
-    return new EventDatabase($queue);
-}, ['queue']);
+Server::setResource('consumer', function (Group $pools) {
+    return $pools->get('consumer')->pop()->getResource();
+}, ['pools']);
 
-Server::setResource('queueForMessaging', function (Connection $queue) {
-    return new Messaging($queue);
-}, ['queue']);
+Server::setResource('queueForUsage', function (Publisher $publisher) {
+    return new Usage($publisher);
+}, ['publisher']);
 
-Server::setResource('queueForMails', function (Connection $queue) {
-    return new Mail($queue);
-}, ['queue']);
+Server::setResource('queueForUsageDump', function (Publisher $publisher) {
+    return new UsageDump($publisher);
+}, ['publisher']);
 
-Server::setResource('queueForBuilds', function (Connection $queue) {
-    return new Build($queue);
-}, ['queue']);
+Server::setResource('queueForDatabase', function (Publisher $publisher) {
+    return new EventDatabase($publisher);
+}, ['publisher']);
 
-Server::setResource('queueForDeletes', function (Connection $queue) {
-    return new Delete($queue);
-}, ['queue']);
+Server::setResource('queueForMessaging', function (Publisher $publisher) {
+    return new Messaging($publisher);
+}, ['publisher']);
 
-Server::setResource('queueForEvents', function (Connection $queue) {
-    return new Event($queue);
-}, ['queue']);
+Server::setResource('queueForMails', function (Publisher $publisher) {
+    return new Mail($publisher);
+}, ['publisher']);
 
-Server::setResource('queueForAudits', function (Connection $queue) {
-    return new Audit($queue);
-}, ['queue']);
+Server::setResource('queueForBuilds', function (Publisher $publisher) {
+    return new Build($publisher);
+}, ['publisher']);
 
-Server::setResource('queueForFunctions', function (Connection $queue) {
-    return new Func($queue);
-}, ['queue']);
+Server::setResource('queueForDeletes', function (Publisher $publisher) {
+    return new Delete($publisher);
+}, ['publisher']);
 
-Server::setResource('queueForCertificates', function (Connection $queue) {
-    return new Certificate($queue);
-}, ['queue']);
+Server::setResource('queueForEvents', function (Publisher $publisher) {
+    return new Event($publisher);
+}, ['publisher']);
 
-Server::setResource('queueForMigrations', function (Connection $queue) {
-    return new Migration($queue);
-}, ['queue']);
+Server::setResource('queueForAudits', function (Publisher $publisher) {
+    return new Audit($publisher);
+}, ['publisher']);
+
+Server::setResource('queueForFunctions', function (Publisher $publisher) {
+    return new Func($publisher);
+}, ['publisher']);
+
+Server::setResource('queueForCertificates', function (Publisher $publisher) {
+    return new Certificate($publisher);
+}, ['publisher']);
+
+Server::setResource('queueForMigrations', function (Publisher $publisher) {
+    return new Migration($publisher);
+}, ['publisher']);
 
 Server::setResource('logger', function (Registry $register) {
     return $register->get('logger');
@@ -386,7 +390,7 @@ try {
      */
     $platform->init(Service::TYPE_WORKER, [
         'workersNum' => System::getEnv('_APP_WORKERS_NUM', 1),
-        'connection' => $pools->get('queue')->pop()->getResource(),
+        'connection' => $pools->get('consumer')->pop()->getResource(),
         'workerName' => strtolower($workerName) ?? null,
         'queueName' => $queueName
     ]);
diff --git a/composer.json b/composer.json
index 8f5bb54f79..fe3400cfe5 100644
--- a/composer.json
+++ b/composer.json
@@ -45,13 +45,13 @@
         "ext-sockets": "*",
         "appwrite/php-runtimes": "0.16.*",
         "appwrite/php-clamav": "2.0.*",
-        "utopia-php/abuse": "0.47.*",
+        "utopia-php/abuse": "0.49.*",
         "utopia-php/analytics": "0.10.*",
-        "utopia-php/audit": "0.47.*",
+        "utopia-php/audit": "0.49.*",
         "utopia-php/cache": "0.11.*",
         "utopia-php/cli": "0.15.*",
         "utopia-php/config": "0.2.*",
-        "utopia-php/database": "0.56.4",
+        "utopia-php/database": "0.58.4",
         "utopia-php/domains": "0.5.*",
         "utopia-php/dsn": "0.2.1",
         "utopia-php/framework": "0.33.*",
@@ -62,10 +62,10 @@
         "utopia-php/messaging": "0.14.*",
         "utopia-php/migration": "0.6.*",
         "utopia-php/orchestration": "0.9.*",
-        "utopia-php/platform": "0.7.1",
+        "utopia-php/platform": "0.7.3",
         "utopia-php/pools": "0.5.*",
         "utopia-php/preloader": "0.2.*",
-        "utopia-php/queue": "0.7.*",
+        "utopia-php/queue": "0.8.*",
         "utopia-php/registry": "0.5.*",
         "utopia-php/storage": "0.18.*",
         "utopia-php/swoole": "0.8.*",
diff --git a/composer.lock b/composer.lock
index a65fabb429..87ee9979ae 100644
--- a/composer.lock
+++ b/composer.lock
@@ -4,7 +4,7 @@
         "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies",
         "This file is @generated automatically"
     ],
-    "content-hash": "e8d26e7e836db255ba42cf55c3798c97",
+    "content-hash": "232691925e05350c7a3831a4e43d79d1",
     "packages": [
         {
             "name": "adhocore/jwt",
@@ -1705,6 +1705,137 @@
             },
             "time": "2024-05-08T12:18:48+00:00"
         },
+        {
+            "name": "paragonie/random_compat",
+            "version": "v9.99.100",
+            "source": {
+                "type": "git",
+                "url": "https://github.com/paragonie/random_compat.git",
+                "reference": "996434e5492cb4c3edcb9168db6fbb1359ef965a"
+            },
+            "dist": {
+                "type": "zip",
+                "url": "https://api.github.com/repos/paragonie/random_compat/zipball/996434e5492cb4c3edcb9168db6fbb1359ef965a",
+                "reference": "996434e5492cb4c3edcb9168db6fbb1359ef965a",
+                "shasum": ""
+            },
+            "require": {
+                "php": ">= 7"
+            },
+            "require-dev": {
+                "phpunit/phpunit": "4.*|5.*",
+                "vimeo/psalm": "^1"
+            },
+            "suggest": {
+                "ext-libsodium": "Provides a modern crypto API that can be used to generate random bytes."
+            },
+            "type": "library",
+            "notification-url": "https://packagist.org/downloads/",
+            "license": [
+                "MIT"
+            ],
+            "authors": [
+                {
+                    "name": "Paragon Initiative Enterprises",
+                    "email": "security@paragonie.com",
+                    "homepage": "https://paragonie.com"
+                }
+            ],
+            "description": "PHP 5.x polyfill for random_bytes() and random_int() from PHP 7",
+            "keywords": [
+                "csprng",
+                "polyfill",
+                "pseudorandom",
+                "random"
+            ],
+            "support": {
+                "email": "info@paragonie.com",
+                "issues": "https://github.com/paragonie/random_compat/issues",
+                "source": "https://github.com/paragonie/random_compat"
+            },
+            "time": "2020-10-15T08:29:30+00:00"
+        },
+        {
+            "name": "php-amqplib/php-amqplib",
+            "version": "v3.7.2",
+            "source": {
+                "type": "git",
+                "url": "https://github.com/php-amqplib/php-amqplib.git",
+                "reference": "738a73eb0019b6c99d9bc25d7a0c0dd8f56a5199"
+            },
+            "dist": {
+                "type": "zip",
+                "url": "https://api.github.com/repos/php-amqplib/php-amqplib/zipball/738a73eb0019b6c99d9bc25d7a0c0dd8f56a5199",
+                "reference": "738a73eb0019b6c99d9bc25d7a0c0dd8f56a5199",
+                "shasum": ""
+            },
+            "require": {
+                "ext-mbstring": "*",
+                "ext-sockets": "*",
+                "php": "^7.2||^8.0",
+                "phpseclib/phpseclib": "^2.0|^3.0"
+            },
+            "conflict": {
+                "php": "7.4.0 - 7.4.1"
+            },
+            "replace": {
+                "videlalvaro/php-amqplib": "self.version"
+            },
+            "require-dev": {
+                "ext-curl": "*",
+                "nategood/httpful": "^0.2.20",
+                "phpunit/phpunit": "^7.5|^9.5",
+                "squizlabs/php_codesniffer": "^3.6"
+            },
+            "type": "library",
+            "extra": {
+                "branch-alias": {
+                    "dev-master": "3.0-dev"
+                }
+            },
+            "autoload": {
+                "psr-4": {
+                    "PhpAmqpLib\\": "PhpAmqpLib/"
+                }
+            },
+            "notification-url": "https://packagist.org/downloads/",
+            "license": [
+                "LGPL-2.1-or-later"
+            ],
+            "authors": [
+                {
+                    "name": "Alvaro Videla",
+                    "role": "Original Maintainer"
+                },
+                {
+                    "name": "Raúl Araya",
+                    "email": "nubeiro@gmail.com",
+                    "role": "Maintainer"
+                },
+                {
+                    "name": "Luke Bakken",
+                    "email": "luke@bakken.io",
+                    "role": "Maintainer"
+                },
+                {
+                    "name": "Ramūnas Dronga",
+                    "email": "github@ramuno.lt",
+                    "role": "Maintainer"
+                }
+            ],
+            "description": "Formerly videlalvaro/php-amqplib.  This library is a pure PHP implementation of the AMQP protocol. It's been tested against RabbitMQ.",
+            "homepage": "https://github.com/php-amqplib/php-amqplib/",
+            "keywords": [
+                "message",
+                "queue",
+                "rabbitmq"
+            ],
+            "support": {
+                "issues": "https://github.com/php-amqplib/php-amqplib/issues",
+                "source": "https://github.com/php-amqplib/php-amqplib/tree/v3.7.2"
+            },
+            "time": "2024-11-21T09:21:41+00:00"
+        },
         {
             "name": "php-http/discovery",
             "version": "1.20.0",
@@ -1865,6 +1996,116 @@
             ],
             "time": "2023-11-25T22:23:28+00:00"
         },
+        {
+            "name": "phpseclib/phpseclib",
+            "version": "3.0.43",
+            "source": {
+                "type": "git",
+                "url": "https://github.com/phpseclib/phpseclib.git",
+                "reference": "709ec107af3cb2f385b9617be72af8cf62441d02"
+            },
+            "dist": {
+                "type": "zip",
+                "url": "https://api.github.com/repos/phpseclib/phpseclib/zipball/709ec107af3cb2f385b9617be72af8cf62441d02",
+                "reference": "709ec107af3cb2f385b9617be72af8cf62441d02",
+                "shasum": ""
+            },
+            "require": {
+                "paragonie/constant_time_encoding": "^1|^2|^3",
+                "paragonie/random_compat": "^1.4|^2.0|^9.99.99",
+                "php": ">=5.6.1"
+            },
+            "require-dev": {
+                "phpunit/phpunit": "*"
+            },
+            "suggest": {
+                "ext-dom": "Install the DOM extension to load XML formatted public keys.",
+                "ext-gmp": "Install the GMP (GNU Multiple Precision) extension in order to speed up arbitrary precision integer arithmetic operations.",
+                "ext-libsodium": "SSH2/SFTP can make use of some algorithms provided by the libsodium-php extension.",
+                "ext-mcrypt": "Install the Mcrypt extension in order to speed up a few other cryptographic operations.",
+                "ext-openssl": "Install the OpenSSL extension in order to speed up a wide variety of cryptographic operations."
+            },
+            "type": "library",
+            "autoload": {
+                "files": [
+                    "phpseclib/bootstrap.php"
+                ],
+                "psr-4": {
+                    "phpseclib3\\": "phpseclib/"
+                }
+            },
+            "notification-url": "https://packagist.org/downloads/",
+            "license": [
+                "MIT"
+            ],
+            "authors": [
+                {
+                    "name": "Jim Wigginton",
+                    "email": "terrafrost@php.net",
+                    "role": "Lead Developer"
+                },
+                {
+                    "name": "Patrick Monnerat",
+                    "email": "pm@datasphere.ch",
+                    "role": "Developer"
+                },
+                {
+                    "name": "Andreas Fischer",
+                    "email": "bantu@phpbb.com",
+                    "role": "Developer"
+                },
+                {
+                    "name": "Hans-Jürgen Petrich",
+                    "email": "petrich@tronic-media.com",
+                    "role": "Developer"
+                },
+                {
+                    "name": "Graham Campbell",
+                    "email": "graham@alt-three.com",
+                    "role": "Developer"
+                }
+            ],
+            "description": "PHP Secure Communications Library - Pure-PHP implementations of RSA, AES, SSH2, SFTP, X.509 etc.",
+            "homepage": "http://phpseclib.sourceforge.net",
+            "keywords": [
+                "BigInteger",
+                "aes",
+                "asn.1",
+                "asn1",
+                "blowfish",
+                "crypto",
+                "cryptography",
+                "encryption",
+                "rsa",
+                "security",
+                "sftp",
+                "signature",
+                "signing",
+                "ssh",
+                "twofish",
+                "x.509",
+                "x509"
+            ],
+            "support": {
+                "issues": "https://github.com/phpseclib/phpseclib/issues",
+                "source": "https://github.com/phpseclib/phpseclib/tree/3.0.43"
+            },
+            "funding": [
+                {
+                    "url": "https://github.com/terrafrost",
+                    "type": "github"
+                },
+                {
+                    "url": "https://www.patreon.com/phpseclib",
+                    "type": "patreon"
+                },
+                {
+                    "url": "https://tidelift.com/funding/github/packagist/phpseclib/phpseclib",
+                    "type": "tidelift"
+                }
+            ],
+            "time": "2024-12-14T21:12:59+00:00"
+        },
         {
             "name": "psr/container",
             "version": "2.0.2",
@@ -3136,16 +3377,16 @@
         },
         {
             "name": "utopia-php/abuse",
-            "version": "0.47.0",
+            "version": "0.49.0",
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/abuse.git",
-                "reference": "2b52bb362234d4072b647ed57db1b3be030f57c2"
+                "reference": "76612c274b895aa3d4d1fa27557a6402463eea99"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/abuse/zipball/2b52bb362234d4072b647ed57db1b3be030f57c2",
-                "reference": "2b52bb362234d4072b647ed57db1b3be030f57c2",
+                "url": "https://api.github.com/repos/utopia-php/abuse/zipball/76612c274b895aa3d4d1fa27557a6402463eea99",
+                "reference": "76612c274b895aa3d4d1fa27557a6402463eea99",
                 "shasum": ""
             },
             "require": {
@@ -3153,13 +3394,13 @@
                 "ext-pdo": "*",
                 "ext-redis": "*",
                 "php": ">=8.0",
-                "utopia-php/database": "0.56.*"
+                "utopia-php/database": "0.58.*"
             },
             "require-dev": {
-                "laravel/pint": "1.5.*",
-                "phpbench/phpbench": "^1.2",
-                "phpstan/phpstan": "^1.9",
-                "phpunit/phpunit": "^9.4"
+                "laravel/pint": "1.*",
+                "phpbench/phpbench": "1.*",
+                "phpstan/phpstan": "1.*",
+                "phpunit/phpunit": "9.*"
             },
             "type": "library",
             "autoload": {
@@ -3181,9 +3422,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/abuse/issues",
-                "source": "https://github.com/utopia-php/abuse/tree/0.47.0"
+                "source": "https://github.com/utopia-php/abuse/tree/0.49.0"
             },
-            "time": "2025-01-15T02:41:02+00:00"
+            "time": "2025-02-04T07:33:59+00:00"
         },
         {
             "name": "utopia-php/analytics",
@@ -3233,26 +3474,26 @@
         },
         {
             "name": "utopia-php/audit",
-            "version": "0.47.0",
+            "version": "0.49.0",
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/audit.git",
-                "reference": "1ebd5784ba68645073426f2f04a67726a1bde4d7"
+                "reference": "9d5c5e0cf0f6d9157b911fc3971da4331d71c96d"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/audit/zipball/1ebd5784ba68645073426f2f04a67726a1bde4d7",
-                "reference": "1ebd5784ba68645073426f2f04a67726a1bde4d7",
+                "url": "https://api.github.com/repos/utopia-php/audit/zipball/9d5c5e0cf0f6d9157b911fc3971da4331d71c96d",
+                "reference": "9d5c5e0cf0f6d9157b911fc3971da4331d71c96d",
                 "shasum": ""
             },
             "require": {
                 "php": ">=8.0",
-                "utopia-php/database": "0.56.*"
+                "utopia-php/database": "0.58.*"
             },
             "require-dev": {
-                "laravel/pint": "1.5.*",
-                "phpstan/phpstan": "^1.8",
-                "phpunit/phpunit": "^9.3"
+                "laravel/pint": "1.*",
+                "phpstan/phpstan": "1.*",
+                "phpunit/phpunit": "9.*"
             },
             "type": "library",
             "autoload": {
@@ -3274,9 +3515,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/audit/issues",
-                "source": "https://github.com/utopia-php/audit/tree/0.47.0"
+                "source": "https://github.com/utopia-php/audit/tree/0.49.0"
             },
-            "time": "2025-01-15T02:40:53+00:00"
+            "time": "2025-02-04T07:27:18+00:00"
         },
         {
             "name": "utopia-php/cache",
@@ -3476,22 +3717,22 @@
         },
         {
             "name": "utopia-php/database",
-            "version": "0.56.4",
+            "version": "0.58.4",
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/database.git",
-                "reference": "240478a60797124a885ceac40046fe47c22415b7"
+                "reference": "ff3fd22e4fe757cc2a78f17169f6dcc45c96d0fe"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/database/zipball/240478a60797124a885ceac40046fe47c22415b7",
-                "reference": "240478a60797124a885ceac40046fe47c22415b7",
+                "url": "https://api.github.com/repos/utopia-php/database/zipball/ff3fd22e4fe757cc2a78f17169f6dcc45c96d0fe",
+                "reference": "ff3fd22e4fe757cc2a78f17169f6dcc45c96d0fe",
                 "shasum": ""
             },
             "require": {
                 "ext-mbstring": "*",
                 "ext-pdo": "*",
-                "php": ">=8.3",
+                "php": ">=8.1",
                 "utopia-php/cache": "0.11.*",
                 "utopia-php/framework": "0.33.*",
                 "utopia-php/mongo": "0.3.*"
@@ -3526,9 +3767,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/database/issues",
-                "source": "https://github.com/utopia-php/database/tree/0.56.4"
+                "source": "https://github.com/utopia-php/database/tree/0.58.4"
             },
-            "time": "2025-01-20T09:22:08+00:00"
+            "time": "2025-02-05T02:51:02+00:00"
         },
         {
             "name": "utopia-php/domains",
@@ -3929,35 +4170,35 @@
         },
         {
             "name": "utopia-php/migration",
-            "version": "0.6.15",
+            "version": "0.6.17",
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/migration.git",
-                "reference": "e849ec3e7ad38f5f5273ebb0132b112639cdf01c"
+                "reference": "677a5c4688d7f54d1631a91f76a35d51346cf96b"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/migration/zipball/e849ec3e7ad38f5f5273ebb0132b112639cdf01c",
-                "reference": "e849ec3e7ad38f5f5273ebb0132b112639cdf01c",
+                "url": "https://api.github.com/repos/utopia-php/migration/zipball/677a5c4688d7f54d1631a91f76a35d51346cf96b",
+                "reference": "677a5c4688d7f54d1631a91f76a35d51346cf96b",
                 "shasum": ""
             },
             "require": {
-                "appwrite/appwrite": "11.1.*",
+                "appwrite/appwrite": "11.*",
                 "ext-curl": "*",
                 "ext-openssl": "*",
-                "php": "8.3.*",
-                "utopia-php/database": "0.56.*",
+                "php": ">=8.1",
+                "utopia-php/database": "0.58.*",
                 "utopia-php/dsn": "0.2.*",
                 "utopia-php/framework": "0.33.*",
                 "utopia-php/storage": "0.18.*"
             },
             "require-dev": {
                 "ext-pdo": "*",
-                "laravel/pint": "1.17.*",
-                "phpstan/phpstan": "1.11.*",
-                "phpunit/phpunit": "11.2.*",
+                "laravel/pint": "1.*",
+                "phpstan/phpstan": "1.*",
+                "phpunit/phpunit": "11.*",
                 "utopia-php/cli": "0.16.*",
-                "vlucas/phpdotenv": "5.6.*"
+                "vlucas/phpdotenv": "5.*"
             },
             "type": "library",
             "autoload": {
@@ -3979,9 +4220,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/migration/issues",
-                "source": "https://github.com/utopia-php/migration/tree/0.6.15"
+                "source": "https://github.com/utopia-php/migration/tree/0.6.17"
             },
-            "time": "2025-01-15T04:55:08+00:00"
+            "time": "2025-02-05T05:27:29+00:00"
         },
         {
             "name": "utopia-php/mongo",
@@ -4095,16 +4336,16 @@
         },
         {
             "name": "utopia-php/platform",
-            "version": "0.7.1",
+            "version": "0.7.3",
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/platform.git",
-                "reference": "3433a0f1a54988f2a59c735f507745cb2c24638a"
+                "reference": "463c2d817c893d7dbb678c2eac7a8291f2710e25"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/platform/zipball/3433a0f1a54988f2a59c735f507745cb2c24638a",
-                "reference": "3433a0f1a54988f2a59c735f507745cb2c24638a",
+                "url": "https://api.github.com/repos/utopia-php/platform/zipball/463c2d817c893d7dbb678c2eac7a8291f2710e25",
+                "reference": "463c2d817c893d7dbb678c2eac7a8291f2710e25",
                 "shasum": ""
             },
             "require": {
@@ -4113,7 +4354,7 @@
                 "php": ">=8.0",
                 "utopia-php/cli": "0.15.*",
                 "utopia-php/framework": "0.33.*",
-                "utopia-php/queue": "0.7.*"
+                "utopia-php/queue": "0.8.*"
             },
             "require-dev": {
                 "laravel/pint": "1.2.*",
@@ -4139,9 +4380,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/platform/issues",
-                "source": "https://github.com/utopia-php/platform/tree/0.7.1"
+                "source": "https://github.com/utopia-php/platform/tree/0.7.3"
             },
-            "time": "2024-10-22T10:27:49+00:00"
+            "time": "2025-02-04T15:09:00+00:00"
         },
         {
             "name": "utopia-php/pools",
@@ -4249,25 +4490,28 @@
         },
         {
             "name": "utopia-php/queue",
-            "version": "0.7.3",
+            "version": "0.8.6",
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/queue.git",
-                "reference": "16074a98ee7d6212bc1228de200e13db470c098a"
+                "reference": "b713b997285c29d120bbcbe3d6e93762d850f87c"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/queue/zipball/16074a98ee7d6212bc1228de200e13db470c098a",
-                "reference": "16074a98ee7d6212bc1228de200e13db470c098a",
+                "url": "https://api.github.com/repos/utopia-php/queue/zipball/b713b997285c29d120bbcbe3d6e93762d850f87c",
+                "reference": "b713b997285c29d120bbcbe3d6e93762d850f87c",
                 "shasum": ""
             },
             "require": {
-                "php": ">=8.1",
+                "php": ">=8.3",
+                "php-amqplib/php-amqplib": "^3.7",
                 "utopia-php/cli": "0.15.*",
-                "utopia-php/framework": "0.*.*",
+                "utopia-php/fetch": "^0.3.0",
+                "utopia-php/framework": "0.33.*",
                 "utopia-php/telemetry": "0.1.*"
             },
             "require-dev": {
+                "ext-redis": "*",
                 "laravel/pint": "^0.2.3",
                 "phpstan/phpstan": "^1.8",
                 "phpunit/phpunit": "^9.5.5",
@@ -4305,9 +4549,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/queue/issues",
-                "source": "https://github.com/utopia-php/queue/tree/0.7.3"
+                "source": "https://github.com/utopia-php/queue/tree/0.8.6"
             },
-            "time": "2024-11-13T12:47:48+00:00"
+            "time": "2025-02-10T03:35:00+00:00"
         },
         {
             "name": "utopia-php/registry",
diff --git a/src/Appwrite/Auth/OAuth2/Slack.php b/src/Appwrite/Auth/OAuth2/Slack.php
index 8898f4d1f7..9c87e45ed6 100644
--- a/src/Appwrite/Auth/OAuth2/Slack.php
+++ b/src/Appwrite/Auth/OAuth2/Slack.php
@@ -20,10 +20,9 @@ class Slack extends OAuth2
      * @var array
      */
     protected array $scopes = [
-        'identity.avatar',
-        'identity.basic',
-        'identity.email',
-        'identity.team'
+        'openid',
+        'email',
+        'profile'
     ];
 
     /**
@@ -35,14 +34,15 @@ class Slack extends OAuth2
     }
 
     /**
+     * @link https://api.slack.com/authentication/oauth-v2
+     *
      * @return string
      */
     public function getLoginURL(): string
     {
-        // https://api.slack.com/docs/oauth#step_1_-_sending_users_to_authorize_and_or_install
-        return 'https://slack.com/oauth/authorize?' . \http_build_query([
+        return 'https://slack.com/oauth/v2/authorize?' . \http_build_query([
             'client_id' => $this->appID,
-            'scope' => \implode(' ', $this->getScopes()),
+            'user_scope' => \implode(' ', $this->getScopes()),
             'redirect_uri' => $this->callback,
             'state' => \json_encode($this->state)
         ]);
@@ -56,16 +56,15 @@ class Slack extends OAuth2
     protected function getTokens(string $code): array
     {
         if (empty($this->tokens)) {
-            // https://api.slack.com/docs/oauth#step_3_-_exchanging_a_verification_code_for_an_access_token
             $this->tokens = \json_decode($this->request(
                 'GET',
-                'https://slack.com/api/oauth.access?' . \http_build_query([
+                'https://slack.com/api/oauth.v2.access?' . \http_build_query([
                     'client_id' => $this->appID,
                     'client_secret' => $this->appSecret,
                     'code' => $code,
                     'redirect_uri' => $this->callback
                 ])
-            ), true);
+            ), true)['authed_user'] ?? [];
         }
 
         return $this->tokens;
@@ -80,13 +79,13 @@ class Slack extends OAuth2
     {
         $this->tokens = \json_decode($this->request(
             'GET',
-            'https://slack.com/api/oauth.access?' . \http_build_query([
+            'https://slack.com/api/oauth.v2.access?' . \http_build_query([
                 'client_id' => $this->appID,
                 'client_secret' => $this->appSecret,
                 'refresh_token' => $refreshToken,
                 'grant_type' => 'refresh_token'
             ])
-        ), true);
+        ), true)['authed_user'] ?? [];
 
         if (empty($this->tokens['refresh_token'])) {
             $this->tokens['refresh_token'] = $refreshToken;
@@ -161,9 +160,9 @@ class Slack extends OAuth2
         if (empty($this->user)) {
             $user = $this->request(
                 'GET',
-                'https://slack.com/api/users.identity?token=' . \urlencode($accessToken)
+                'https://slack.com/api/users.identity',
+                ['Authorization: Bearer ' . \urlencode($accessToken)]
             );
-
             $this->user = \json_decode($user, true);
         }
 
diff --git a/src/Appwrite/Event/Audit.php b/src/Appwrite/Event/Audit.php
index 4b9aa9f5c5..6c2a9c3086 100644
--- a/src/Appwrite/Event/Audit.php
+++ b/src/Appwrite/Event/Audit.php
@@ -2,7 +2,7 @@
 
 namespace Appwrite\Event;
 
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
 
 class Audit extends Event
 {
@@ -12,9 +12,9 @@ class Audit extends Event
     protected string $ip = '';
     protected string $hostname = '';
 
-    public function __construct(protected Connection $connection)
+    public function __construct(protected Publisher $publisher)
     {
-        parent::__construct($connection);
+        parent::__construct($publisher);
 
         $this
             ->setQueue(Event::AUDITS_QUEUE_NAME)
diff --git a/src/Appwrite/Event/Build.php b/src/Appwrite/Event/Build.php
index 831adf8e41..9ea163174f 100644
--- a/src/Appwrite/Event/Build.php
+++ b/src/Appwrite/Event/Build.php
@@ -3,7 +3,7 @@
 namespace Appwrite\Event;
 
 use Utopia\Database\Document;
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
 
 class Build extends Event
 {
@@ -12,9 +12,9 @@ class Build extends Event
     protected ?Document $deployment = null;
     protected ?Document $template = null;
 
-    public function __construct(protected Connection $connection)
+    public function __construct(protected Publisher $publisher)
     {
-        parent::__construct($connection);
+        parent::__construct($publisher);
 
         $this
             ->setQueue(Event::BUILDS_QUEUE_NAME)
diff --git a/src/Appwrite/Event/Certificate.php b/src/Appwrite/Event/Certificate.php
index 6a395417ed..827472ae37 100644
--- a/src/Appwrite/Event/Certificate.php
+++ b/src/Appwrite/Event/Certificate.php
@@ -3,16 +3,16 @@
 namespace Appwrite\Event;
 
 use Utopia\Database\Document;
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
 
 class Certificate extends Event
 {
     protected bool $skipRenewCheck = false;
     protected ?Document $domain = null;
 
-    public function __construct(protected Connection $connection)
+    public function __construct(protected Publisher $publisher)
     {
-        parent::__construct($connection);
+        parent::__construct($publisher);
 
         $this
             ->setQueue(Event::CERTIFICATES_QUEUE_NAME)
diff --git a/src/Appwrite/Event/Database.php b/src/Appwrite/Event/Database.php
index 24123de6c1..d2f70dddf2 100644
--- a/src/Appwrite/Event/Database.php
+++ b/src/Appwrite/Event/Database.php
@@ -4,7 +4,7 @@ namespace Appwrite\Event;
 
 use Utopia\Database\Document;
 use Utopia\DSN\DSN;
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
 
 class Database extends Event
 {
@@ -13,9 +13,9 @@ class Database extends Event
     protected ?Document $collection = null;
     protected ?Document $document = null;
 
-    public function __construct(protected Connection $connection)
+    public function __construct(protected Publisher $publisher)
     {
-        parent::__construct($connection);
+        parent::__construct($publisher);
 
         $this->setClass(Event::DATABASE_CLASS_NAME);
     }
diff --git a/src/Appwrite/Event/Delete.php b/src/Appwrite/Event/Delete.php
index f0af20f21b..450be306d7 100644
--- a/src/Appwrite/Event/Delete.php
+++ b/src/Appwrite/Event/Delete.php
@@ -3,7 +3,7 @@
 namespace Appwrite\Event;
 
 use Utopia\Database\Document;
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
 
 class Delete extends Event
 {
@@ -15,9 +15,9 @@ class Delete extends Event
     protected ?string $hourlyUsageRetentionDatetime = null;
 
 
-    public function __construct(protected Connection $connection)
+    public function __construct(protected Publisher $publisher)
     {
-        parent::__construct($connection);
+        parent::__construct($publisher);
 
         $this
             ->setQueue(Event::DELETE_QUEUE_NAME)
diff --git a/src/Appwrite/Event/Event.php b/src/Appwrite/Event/Event.php
index 5cd5f8e7d6..f56aeeb757 100644
--- a/src/Appwrite/Event/Event.php
+++ b/src/Appwrite/Event/Event.php
@@ -4,8 +4,8 @@ namespace Appwrite\Event;
 
 use InvalidArgumentException;
 use Utopia\Database\Document;
-use Utopia\Queue\Client;
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
+use Utopia\Queue\Queue;
 
 class Event
 {
@@ -58,10 +58,10 @@ class Event
     protected bool $paused = false;
 
     /**
-     * @param Connection $connection
+     * @param Publisher $publisher
      * @return void
      */
-    public function __construct(protected Connection $connection)
+    public function __construct(protected Publisher $publisher)
     {
     }
 
@@ -345,12 +345,11 @@ class Event
         }
 
         /** The getter is required since events like Databases need to override the queue name depending on the project */
-        $client = new Client($this->getQueue(), $this->connection);
+        $queue = new Queue($this->getQueue());
 
         // Merge the base payload with any trimmed values
         $payload = array_merge($this->preparePayload(), $this->trimPayload());
-
-        return $client->enqueue($payload);
+        return $this->publisher->enqueue($queue, $payload);
     }
 
     /**
diff --git a/src/Appwrite/Event/Func.php b/src/Appwrite/Event/Func.php
index b3945fccb8..ae316c84e5 100644
--- a/src/Appwrite/Event/Func.php
+++ b/src/Appwrite/Event/Func.php
@@ -3,7 +3,7 @@
 namespace Appwrite\Event;
 
 use Utopia\Database\Document;
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
 
 class Func extends Event
 {
@@ -19,9 +19,9 @@ class Func extends Event
     protected ?Document $function = null;
     protected ?Document $execution = null;
 
-    public function __construct(protected Connection $connection)
+    public function __construct(protected Publisher $publisher)
     {
-        parent::__construct($connection);
+        parent::__construct($publisher);
 
         $this
             ->setQueue(Event::FUNCTIONS_QUEUE_NAME)
diff --git a/src/Appwrite/Event/Mail.php b/src/Appwrite/Event/Mail.php
index 1c9e539cdb..87312182ea 100644
--- a/src/Appwrite/Event/Mail.php
+++ b/src/Appwrite/Event/Mail.php
@@ -2,7 +2,7 @@
 
 namespace Appwrite\Event;
 
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
 
 class Mail extends Event
 {
@@ -15,9 +15,9 @@ class Mail extends Event
     protected string $bodyTemplate = '';
     protected array $attachment = [];
 
-    public function __construct(protected Connection $connection)
+    public function __construct(protected Publisher $publisher)
     {
-        parent::__construct($connection);
+        parent::__construct($publisher);
 
         $this
             ->setQueue(Event::MAILS_QUEUE_NAME)
diff --git a/src/Appwrite/Event/Messaging.php b/src/Appwrite/Event/Messaging.php
index 61dbe9c427..3ddbac1040 100644
--- a/src/Appwrite/Event/Messaging.php
+++ b/src/Appwrite/Event/Messaging.php
@@ -3,7 +3,7 @@
 namespace Appwrite\Event;
 
 use Utopia\Database\Document;
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
 
 class Messaging extends Event
 {
@@ -14,9 +14,9 @@ class Messaging extends Event
     protected ?string $scheduledAt = null;
     protected ?string $providerType = null;
 
-    public function __construct(protected Connection $connection)
+    public function __construct(protected Publisher $publisher)
     {
-        parent::__construct($connection);
+        parent::__construct($publisher);
 
         $this
             ->setQueue(Event::MESSAGING_QUEUE_NAME)
diff --git a/src/Appwrite/Event/Migration.php b/src/Appwrite/Event/Migration.php
index 5fb2d5a106..bbb8d77c73 100644
--- a/src/Appwrite/Event/Migration.php
+++ b/src/Appwrite/Event/Migration.php
@@ -3,16 +3,16 @@
 namespace Appwrite\Event;
 
 use Utopia\Database\Document;
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
 
 class Migration extends Event
 {
     protected string $type = '';
     protected ?Document $migration = null;
 
-    public function __construct(protected Connection $connection)
+    public function __construct(protected Publisher $publisher)
     {
-        parent::__construct($connection);
+        parent::__construct($publisher);
 
         $this
             ->setQueue(Event::MIGRATIONS_QUEUE_NAME)
diff --git a/src/Appwrite/Event/Usage.php b/src/Appwrite/Event/Usage.php
index 5609859f37..c70cea5c73 100644
--- a/src/Appwrite/Event/Usage.php
+++ b/src/Appwrite/Event/Usage.php
@@ -3,16 +3,16 @@
 namespace Appwrite\Event;
 
 use Utopia\Database\Document;
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
 
 class Usage extends Event
 {
     protected array $metrics = [];
     protected array $reduce  = [];
 
-    public function __construct(protected Connection $connection)
+    public function __construct(protected Publisher $publisher)
     {
-        parent::__construct($connection);
+        parent::__construct($publisher);
 
         $this
             ->setQueue(Event::USAGE_QUEUE_NAME)
diff --git a/src/Appwrite/Event/UsageDump.php b/src/Appwrite/Event/UsageDump.php
index 6f44de4eda..a70716e94f 100644
--- a/src/Appwrite/Event/UsageDump.php
+++ b/src/Appwrite/Event/UsageDump.php
@@ -2,15 +2,15 @@
 
 namespace Appwrite\Event;
 
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
 
 class UsageDump extends Event
 {
     protected array $stats;
 
-    public function __construct(protected Connection $connection)
+    public function __construct(protected Publisher $publisher)
     {
-        parent::__construct($connection);
+        parent::__construct($publisher);
 
         $this
             ->setQueue(Event::USAGE_DUMP_QUEUE_NAME)
diff --git a/src/Appwrite/Event/Webhook.php b/src/Appwrite/Event/Webhook.php
index 3e0dbe446f..5cc65758ee 100644
--- a/src/Appwrite/Event/Webhook.php
+++ b/src/Appwrite/Event/Webhook.php
@@ -2,13 +2,13 @@
 
 namespace Appwrite\Event;
 
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
 
 class Webhook extends Event
 {
-    public function __construct(protected Connection $connection)
+    public function __construct(protected Publisher $publisher)
     {
-        parent::__construct($connection);
+        parent::__construct($publisher);
 
         $this
             ->setQueue(Event::WEBHOOK_QUEUE_NAME)
diff --git a/src/Appwrite/Platform/Services/Tasks.php b/src/Appwrite/Platform/Services/Tasks.php
index 7a0d5b60ac..c09f961fc0 100644
--- a/src/Appwrite/Platform/Services/Tasks.php
+++ b/src/Appwrite/Platform/Services/Tasks.php
@@ -6,7 +6,6 @@ use Appwrite\Platform\Tasks\Doctor;
 use Appwrite\Platform\Tasks\Install;
 use Appwrite\Platform\Tasks\Maintenance;
 use Appwrite\Platform\Tasks\Migrate;
-use Appwrite\Platform\Tasks\QueueCount;
 use Appwrite\Platform\Tasks\QueueRetry;
 use Appwrite\Platform\Tasks\ScheduleExecutions;
 use Appwrite\Platform\Tasks\ScheduleFunctions;
@@ -29,7 +28,6 @@ class Tasks extends Service
             ->addAction(Install::getName(), new Install())
             ->addAction(Maintenance::getName(), new Maintenance())
             ->addAction(Migrate::getName(), new Migrate())
-            ->addAction(QueueCount::getName(), new QueueCount())
             ->addAction(QueueRetry::getName(), new QueueRetry())
             ->addAction(SDKs::getName(), new SDKs())
             ->addAction(SSL::getName(), new SSL())
diff --git a/src/Appwrite/Platform/Tasks/QueueCount.php b/src/Appwrite/Platform/Tasks/QueueCount.php
deleted file mode 100644
index b02165c1d2..0000000000
--- a/src/Appwrite/Platform/Tasks/QueueCount.php
+++ /dev/null
@@ -1,57 +0,0 @@
-<?php
-
-namespace Appwrite\Platform\Tasks;
-
-use Utopia\CLI\Console;
-use Utopia\Platform\Action;
-use Utopia\Queue\Client;
-use Utopia\Queue\Connection;
-use Utopia\Validator\Text;
-use Utopia\Validator\WhiteList;
-
-class QueueCount extends Action
-{
-    public static function getName(): string
-    {
-        return 'queue-count';
-    }
-
-
-    public function __construct()
-    {
-        $this
-            ->desc('Return the number of from a specific queue identified by the name parameter with a specific type')
-            ->param('name', '', new Text(100), 'Queue name')
-            ->param('type', '', new WhiteList([
-                'success',
-                'failed',
-                'processing',
-            ]), 'Queue type')
-            ->inject('queue')
-            ->callback(fn ($name, $type, $queue) => $this->action($name, $type, $queue));
-    }
-
-    /**
-     * @param string $name The name of the queue to count the jobs from
-     * @param string $type The type of jobs to count
-     * @param Connection $queue
-     */
-    public function action(string $name, string $type, Connection $queue): void
-    {
-        if (!$name) {
-            Console::error('Missing required parameter $name');
-            return;
-        }
-
-        $queueClient = new Client($name, $queue);
-
-        $count = match ($type) {
-            'success' => $queueClient->countSuccessfulJobs(),
-            'failed' => $queueClient->countFailedJobs(),
-            'processing' => $queueClient->countProcessingJobs(),
-            default => 0
-        };
-
-        Console::log("Queue: '{$name}' has {$count} {$type} jobs.");
-    }
-}
diff --git a/src/Appwrite/Platform/Tasks/QueueRetry.php b/src/Appwrite/Platform/Tasks/QueueRetry.php
index b6139dc177..9fe4aed799 100644
--- a/src/Appwrite/Platform/Tasks/QueueRetry.php
+++ b/src/Appwrite/Platform/Tasks/QueueRetry.php
@@ -4,8 +4,8 @@ namespace Appwrite\Platform\Tasks;
 
 use Utopia\CLI\Console;
 use Utopia\Platform\Action;
-use Utopia\Queue\Client;
-use Utopia\Queue\Connection;
+use Utopia\Queue\Publisher;
+use Utopia\Queue\Queue;
 use Utopia\Validator\Text;
 use Utopia\Validator\Wildcard;
 
@@ -23,33 +23,24 @@ class QueueRetry extends Action
             ->desc('Retry failed jobs from a specific queue identified by the name parameter')
             ->param('name', '', new Text(100), 'Queue name')
             ->param('limit', 0, new Wildcard(), 'jobs limit', true)
-            ->inject('queue')
-            ->callback(fn ($name, $limit, $queue) => $this->action($name, $limit, $queue));
+            ->inject('publisher')
+            ->callback(fn ($name, $limit, $publisher) => $this->action($name, $limit, $publisher));
     }
 
     /**
      * @param string $name The name of the queue to retry jobs from
      * @param  mixed $limit
-     * @param Connection $queue
+     * @param Publisher $publisher
      */
-    public function action(string $name, mixed $limit, Connection $queue): void
+    public function action(string $name, mixed $limit, Publisher $publisher): void
     {
-
         if (!$name) {
             Console::error('Missing required parameter $name');
             return;
         }
 
         $limit = (int)$limit;
-        $queueClient = new Client($name, $queue);
-
-        if ($queueClient->countFailedJobs() === 0) {
-            Console::error('No failed jobs found.');
-            return;
-        }
-
         Console::log('Retrying failed jobs...');
-
-        $queueClient->retry($limit);
+        $publisher->retry(new Queue($name), $limit);
     }
 }
diff --git a/src/Appwrite/Platform/Tasks/ScheduleExecutions.php b/src/Appwrite/Platform/Tasks/ScheduleExecutions.php
index 086bad513e..7cd76b480d 100644
--- a/src/Appwrite/Platform/Tasks/ScheduleExecutions.php
+++ b/src/Appwrite/Platform/Tasks/ScheduleExecutions.php
@@ -29,7 +29,7 @@ class ScheduleExecutions extends ScheduleBase
 
     protected function enqueueResources(Group $pools, Database $dbForPlatform, callable $getProjectDB): void
     {
-        $queue = $pools->get('queue')->pop();
+        $queue = $pools->get('publisher')->pop();
         $connection = $queue->getResource();
         $queueForFunctions = new Func($connection);
         $intervalEnd = (new \DateTime())->modify('+' . self::ENQUEUE_TIMER . ' seconds');
diff --git a/src/Appwrite/Platform/Tasks/ScheduleFunctions.php b/src/Appwrite/Platform/Tasks/ScheduleFunctions.php
index c443bb6c2d..5b8e3027a7 100644
--- a/src/Appwrite/Platform/Tasks/ScheduleFunctions.php
+++ b/src/Appwrite/Platform/Tasks/ScheduleFunctions.php
@@ -73,7 +73,7 @@ class ScheduleFunctions extends ScheduleBase
             \go(function () use ($delay, $scheduleKeys, $pools, $dbForPlatform) {
                 \sleep($delay); // in seconds
 
-                $queue = $pools->get('queue')->pop();
+                $queue = $pools->get('publisher')->pop();
                 $connection = $queue->getResource();
 
                 foreach ($scheduleKeys as $scheduleKey) {
diff --git a/src/Appwrite/Platform/Tasks/ScheduleMessages.php b/src/Appwrite/Platform/Tasks/ScheduleMessages.php
index 5d997fc5bb..201d5eab53 100644
--- a/src/Appwrite/Platform/Tasks/ScheduleMessages.php
+++ b/src/Appwrite/Platform/Tasks/ScheduleMessages.php
@@ -41,7 +41,7 @@ class ScheduleMessages extends ScheduleBase
             }
 
             \go(function () use ($schedule, $pools, $dbForPlatform) {
-                $queue = $pools->get('queue')->pop();
+                $queue = $pools->get('publisher')->pop();
                 $connection = $queue->getResource();
                 $queueForMessaging = new Messaging($connection);
 
diff --git a/src/Appwrite/Platform/Workers/Certificates.php b/src/Appwrite/Platform/Workers/Certificates.php
index e763cb54ee..7e220b2734 100644
--- a/src/Appwrite/Platform/Workers/Certificates.php
+++ b/src/Appwrite/Platform/Workers/Certificates.php
@@ -370,6 +370,8 @@ class Certificates extends Action
             /** 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())
diff --git a/src/Appwrite/Utopia/Request.php b/src/Appwrite/Utopia/Request.php
index f8c0439293..480fce58b0 100644
--- a/src/Appwrite/Utopia/Request.php
+++ b/src/Appwrite/Utopia/Request.php
@@ -2,8 +2,10 @@
 
 namespace Appwrite\Utopia;
 
+use Appwrite\Auth\Auth;
 use Appwrite\Utopia\Request\Filter;
 use Swoole\Http\Request as SwooleRequest;
+use Utopia\Database\Validator\Authorization;
 use Utopia\Route;
 use Utopia\Swoole\Request as UtopiaRequest;
 
@@ -180,4 +182,27 @@ class Request extends UtopiaRequest
         $headers = $this->getHeaders();
         return $headers[$key] ?? $default;
     }
+
+    /**
+    * Get User Agent
+    *
+    * Method for getting User Agent. Preferring forwarded agent for privileged users; otherwise returns default.
+    *
+    * @param  string  $default
+    * @return string
+    */
+    public function getUserAgent(string $default = ''): string
+    {
+        $forwardedUserAgent = $this->getHeader('x-forwarded-user-agent');
+        if (!empty($forwardedUserAgent)) {
+            $roles = Authorization::getRoles();
+            $isAppUser = Auth::isAppUser($roles);
+
+            if ($isAppUser) {
+                return $forwardedUserAgent;
+            }
+        }
+
+        return UtopiaRequest::getUserAgent($default);
+    }
 }
diff --git a/tests/e2e/Services/Account/AccountCustomClientTest.php b/tests/e2e/Services/Account/AccountCustomClientTest.php
index cca27cc3be..788f949fb3 100644
--- a/tests/e2e/Services/Account/AccountCustomClientTest.php
+++ b/tests/e2e/Services/Account/AccountCustomClientTest.php
@@ -2307,6 +2307,60 @@ class AccountCustomClientTest extends Scope
         $this->assertNotEmpty($response['body']['$id']);
         $this->assertNotEmpty($response['body']['expire']);
         $this->assertEmpty($response['body']['secret']);
+        $this->assertEquals('browser', $response['body']['clientType']);
+        $this->assertEquals('CH', $response['body']['clientCode']);
+        $this->assertEquals('Chrome', $response['body']['clientName']);
+
+        // Forwarded User Agent with API Key
+        $response = $this->client->call(Client::METHOD_POST, '/users/' . $data['id'] . '/tokens', [
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ], [
+            'expire' => 60
+        ]);
+
+        $userId = $response['body']['userId'];
+        $secret = $response['body']['secret'];
+
+        $response = $this->client->call(Client::METHOD_POST, '/account/sessions/token', [
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+            'x-forwarded-user-agent' => 'Mozilla/5.0 (Linux; Android 10; K) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Mobile Safari/537.36'
+        ], [
+            'userId' => $userId,
+            'secret' => $secret
+        ]);
+
+        $this->assertEquals('browser', $response['body']['clientType']);
+        $this->assertEquals('CM', actual: $response['body']['clientCode']);
+        $this->assertEquals('Chrome Mobile', $response['body']['clientName']);
+
+        // Forwarded User Agent without API Key
+        $response = $this->client->call(Client::METHOD_POST, '/users/' . $data['id'] . '/tokens', [
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-appwrite-key' => $this->getProject()['apiKey'],
+        ], [
+            'expire' => 60
+        ]);
+
+        $userId = $response['body']['userId'];
+        $secret = $response['body']['secret'];
+
+        $response = $this->client->call(Client::METHOD_POST, '/account/sessions/token', [
+            'content-type' => 'application/json',
+            'x-appwrite-project' => $this->getProject()['$id'],
+            'x-forwarded-user-agent' => 'Mozilla/5.0 (Linux; Android 10; K) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Mobile Safari/537.36'
+        ], [
+            'userId' => $userId,
+            'secret' => $secret
+        ]);
+
+        $this->assertEquals('browser', $response['body']['clientType']);
+        $this->assertEquals('CH', $response['body']['clientCode']);
+        $this->assertEquals('Chrome', $response['body']['clientName']);
 
         /**
          * Test for FAILURE
diff --git a/tests/e2e/Services/GraphQL/FunctionsClientTest.php b/tests/e2e/Services/GraphQL/FunctionsClientTest.php
index e7e8421254..0b43ed5772 100644
--- a/tests/e2e/Services/GraphQL/FunctionsClientTest.php
+++ b/tests/e2e/Services/GraphQL/FunctionsClientTest.php
@@ -2,6 +2,7 @@
 
 namespace Tests\E2E\Services\GraphQL;
 
+use Appwrite\Tests\Async;
 use Tests\E2E\Client;
 use Tests\E2E\Scopes\ProjectCustom;
 use Tests\E2E\Scopes\Scope;
@@ -14,6 +15,7 @@ class FunctionsClientTest extends Scope
     use ProjectCustom;
     use SideClient;
     use Base;
+    use Async;
 
     public function testCreateFunction(): array
     {
@@ -119,7 +121,7 @@ class FunctionsClientTest extends Scope
             ]
         ];
 
-        while (true) {
+        $this->assertEventually(function () use ($projectId, $gqlPayload, &$deployment) {
             $deployment = $this->client->call(Client::METHOD_POST, '/graphql', [
                 'content-type' => 'application/json',
                 'x-appwrite-project' => $projectId,
@@ -130,18 +132,8 @@ class FunctionsClientTest extends Scope
             $this->assertArrayNotHasKey('errors', $deployment['body']);
 
             $deployment = $deployment['body']['data']['functionsGetDeployment'];
-
-            if (
-                $deployment['status'] === 'ready'
-                || $deployment['status'] === 'failed'
-            ) {
-                break;
-            }
-
-            \sleep(1);
-        }
-
-        $this->assertEquals('ready', $deployment['status']);
+            $this->assertEquals('ready', $deployment['status']);
+        });
 
         return $deployment;
     }
diff --git a/tests/e2e/Services/GraphQL/FunctionsServerTest.php b/tests/e2e/Services/GraphQL/FunctionsServerTest.php
index c3606244c4..e49ac43619 100644
--- a/tests/e2e/Services/GraphQL/FunctionsServerTest.php
+++ b/tests/e2e/Services/GraphQL/FunctionsServerTest.php
@@ -2,6 +2,7 @@
 
 namespace Tests\E2E\Services\GraphQL;
 
+use Appwrite\Tests\Async;
 use Tests\E2E\Client;
 use Tests\E2E\Scopes\ProjectCustom;
 use Tests\E2E\Scopes\Scope;
@@ -14,6 +15,7 @@ class FunctionsServerTest extends Scope
     use ProjectCustom;
     use SideServer;
     use Base;
+    use Async;
 
     public function testCreateFunction(): array
     {
@@ -117,7 +119,7 @@ class FunctionsServerTest extends Scope
             ]
         ];
 
-        while (true) {
+        $this->assertEventually(function () use ($projectId, $gqlPayload, &$deployment) {
             $deployment = $this->client->call(Client::METHOD_POST, '/graphql', \array_merge([
                 'content-type' => 'application/json',
                 'x-appwrite-project' => $projectId,
@@ -127,19 +129,8 @@ class FunctionsServerTest extends Scope
             $this->assertArrayNotHasKey('errors', $deployment['body']);
 
             $deployment = $deployment['body']['data']['functionsGetDeployment'];
-
-            if (
-                $deployment['status'] === 'ready'
-                || $deployment['status'] === 'failed'
-            ) {
-                break;
-            }
-
-            \sleep(1);
-        }
-
-        $this->assertEquals('ready', $deployment['status']);
-
+            $this->assertEquals('ready', $deployment['status']);
+        });
         return $deployment;
     }
 
diff --git a/tests/e2e/Services/Health/HealthCustomServerTest.php b/tests/e2e/Services/Health/HealthCustomServerTest.php
index 9d6a04abe6..f2c6a2e5c2 100644
--- a/tests/e2e/Services/Health/HealthCustomServerTest.php
+++ b/tests/e2e/Services/Health/HealthCustomServerTest.php
@@ -67,24 +67,6 @@ class HealthCustomServerTest extends Scope
         return [];
     }
 
-    public function testQueueSuccess(): array
-    {
-        /**
-         * Test for SUCCESS
-         */
-        $response = $this->client->call(Client::METHOD_GET, '/health/queue', array_merge([
-            'content-type' => 'application/json',
-            'x-appwrite-project' => $this->getProject()['$id'],
-        ], $this->getHeaders()), []);
-
-        $this->assertEquals(200, $response['headers']['status-code']);
-        $this->assertEquals('pass', $response['body']['statuses'][0]['status']);
-        $this->assertIsInt($response['body']['statuses'][0]['ping']);
-        $this->assertLessThan(100, $response['body']['statuses'][0]['ping']);
-
-        return [];
-    }
-
     public function testPubSubSuccess(): array
     {
         /**
diff --git a/tests/unit/Event/EventTest.php b/tests/unit/Event/EventTest.php
index 079bb47b65..c852cf2757 100644
--- a/tests/unit/Event/EventTest.php
+++ b/tests/unit/Event/EventTest.php
@@ -5,7 +5,7 @@ namespace Tests\Unit\Event;
 use Appwrite\Event\Event;
 use InvalidArgumentException;
 use PHPUnit\Framework\TestCase;
-use Utopia\Queue\Client;
+use Utopia\Queue\Publisher;
 
 require_once __DIR__ . '/../../../app/init.php';
 
@@ -13,13 +13,14 @@ class EventTest extends TestCase
 {
     protected ?Event $object = null;
     protected string $queue = '';
+    protected Publisher $publisher;
 
     public function setUp(): void
     {
-        global $register;
-        $connection = $register->get('pools')->get('queue')->pop()->getResource();
+        $this->publisher = new MockPublisher();
+
         $this->queue = 'v1-tests' . uniqid();
-        $this->object = new Event($connection);
+        $this->object = new Event($this->publisher);
         $this->object->setClass('TestsV1');
         $this->object->setQueue($this->queue);
     }
@@ -51,10 +52,7 @@ class EventTest extends TestCase
         $this->assertEquals('eventValue1', $this->object->getParam('eventKey1'));
         $this->assertEquals('eventValue2', $this->object->getParam('eventKey2'));
         $this->assertEquals(null, $this->object->getParam('eventKey3'));
-        global $register;
-        $pools = $register->get('pools');
-        $client = new Client($this->object->getQueue(), $pools->get('queue')->pop()->getResource());
-        $this->assertEquals($client->getQueueSize(), 1);
+        $this->assertCount(1, $this->publisher->getEvents($this->object->getQueue()));
     }
 
     public function testReset(): void
diff --git a/tests/unit/Event/MockPublisher.php b/tests/unit/Event/MockPublisher.php
new file mode 100644
index 0000000000..54fcc89358
--- /dev/null
+++ b/tests/unit/Event/MockPublisher.php
@@ -0,0 +1,35 @@
+<?php
+
+namespace Tests\Unit\Event;
+
+use Utopia\Queue\Publisher;
+use Utopia\Queue\Queue;
+
+class MockPublisher implements Publisher
+{
+    private $events = [];
+
+    public function enqueue(Queue $queue, array $payload): bool
+    {
+        if (!isset($this->events[$queue->name])) {
+            $this->events[$queue->name] = [];
+        }
+        $this->events[$queue->name][] = $payload;
+        return true;
+    }
+
+    public function getEvents(string $queue)
+    {
+        return $this->events[$queue] ?? null;
+    }
+
+    public function retry(Queue $queue, int $limit = null): void
+    {
+        // TODO: Implement retry() method.
+    }
+
+    public function getQueueSize(Queue $queue, bool $failedJobs = false): int
+    {
+        return count($this->events[$queue->name]);
+    }
+}
diff --git a/tests/unit/Usage/StatsTest.php b/tests/unit/Usage/StatsTest.php
deleted file mode 100644
index 79fa1f58ec..0000000000
--- a/tests/unit/Usage/StatsTest.php
+++ /dev/null
@@ -1,41 +0,0 @@
-<?php
-
-namespace Tests\Unit\Usage;
-
-use PHPUnit\Framework\TestCase;
-use Utopia\Queue\Client;
-use Utopia\Queue\Connection;
-
-class StatsTest extends TestCase
-{
-    protected ?Connection $connection = null;
-    protected ?Client $client = null;
-
-    protected const QUEUE_NAME = 'usage-test-q';
-
-    public function setUp(): void
-    {
-        global $register;
-        $connection = $register->get('pools')->get('queue')->pop()->getResource();
-        $this->connection = $connection;
-        $this->client     = new Client(self::QUEUE_NAME, $this->connection);
-    }
-
-    public function tearDown(): void
-    {
-    }
-
-    public function testSamePayload(): void
-    {
-        $inToQueue = [
-            'key_1'  => 'value_1',
-            'key_2'  => 'value_2',
-        ];
-
-        $result = $this->client->enqueue($inToQueue);
-        $this->assertTrue($result);
-        $outFromQueue  = $this->connection->leftPopArray('utopia-queue.queue.' . self::QUEUE_NAME, 0)['payload'];
-        $this->assertNotEmpty($outFromQueue);
-        $this->assertSame($inToQueue, $outFromQueue);
-    }
-}