Merge pull request #9703 from appwrite/revert-9659-feat-pool-adapter

Revert "Feat pool adapter"

app/cli.php

@@ -10,13 +10,11 @@ use Appwrite\Event\StatsUsage;
 use Appwrite\Platform\Appwrite;
 use Appwrite\Runtimes\Runtimes;
 use Executor\Executor;
-use Utopia\Cache\Adapter\Pool as CachePool;
 use Utopia\Cache\Adapter\Sharding;
 use Utopia\Cache\Cache;
 use Utopia\CLI\CLI;
 use Utopia\CLI\Console;
 use Utopia\Config\Config;
-use Utopia\Database\Adapter\Pool as DatabasePool;
 use Utopia\Database\Database;
 use Utopia\Database\Document;
 use Utopia\Database\Validator\Authorization;
@@ -24,7 +22,6 @@ use Utopia\DSN\DSN;
 use Utopia\Logger\Log;
 use Utopia\Platform\Service;
 use Utopia\Pools\Group;
-use Utopia\Queue\Broker\Pool as BrokerPool;
 use Utopia\Queue\Publisher;
 use Utopia\Registry\Registry;
 use Utopia\System\System;
@@ -44,7 +41,10 @@ CLI::setResource('cache', function ($pools) {
     $adapters = [];
 
     foreach ($list as $value) {
-        $adapters[] = new CachePool($pools->get($value));
+        $adapters[] = $pools
+            ->get($value)
+            ->pop()
+            ->getResource();
     }
 
     return new Cache(new Sharding($adapters));
@@ -64,8 +64,12 @@ CLI::setResource('dbForPlatform', function ($pools, $cache) {
         $attempts++;
         try {
             // Prepare database connection
-            $adapter = new DatabasePool($pools->get('console'));
-            $dbForPlatform = new Database($adapter, $cache);
+            $dbAdapter = $pools
+                ->get('console')
+                ->pop()
+                ->getResource();
+
+            $dbForPlatform = new Database($dbAdapter, $cache);
 
             $dbForPlatform
                 ->setNamespace('_console')
@@ -83,6 +87,7 @@ CLI::setResource('dbForPlatform', function ($pools, $cache) {
             $ready = true;
         } catch (\Throwable $err) {
             Console::warning($err->getMessage());
+            $pools->get('console')->reclaim();
             sleep($sleep);
         }
     } while ($attempts < $maxAttempts && !$ready);
@@ -132,8 +137,12 @@ CLI::setResource('getProjectDB', function (Group $pools, Database $dbForPlatform
             return $database;
         }
 
-        $adapter = new DatabasePool($pools->get($dsn->getHost()));
-        $database = new Database($adapter, $cache);
+        $dbAdapter = $pools
+            ->get($dsn->getHost())
+            ->pop()
+            ->getResource();
+
+        $database = new Database($dbAdapter, $cache);
         $databases[$dsn->getHost()] = $database;
         $sharedTables = \explode(',', System::getEnv('_APP_DATABASE_SHARED_TABLES', ''));
 
@@ -159,15 +168,21 @@ CLI::setResource('getProjectDB', function (Group $pools, Database $dbForPlatform
 
 CLI::setResource('getLogsDB', function (Group $pools, Cache $cache) {
     $database = null;
-
     return function (?Document $project = null) use ($pools, $cache, $database) {
         if ($database !== null && $project !== null && !$project->isEmpty() && $project->getId() !== 'console') {
             $database->setTenant($project->getInternalId());
             return $database;
         }
 
-        $adapter = new DatabasePool($pools->get('logs'));
-        $database = new Database($adapter, $cache);
+        $dbAdapter = $pools
+            ->get('logs')
+            ->pop()
+            ->getResource();
+
+        $database = new Database(
+            $dbAdapter,
+            $cache
+        );
 
         $database
             ->setSharedTables(true)
@@ -184,14 +199,14 @@ CLI::setResource('getLogsDB', function (Group $pools, Cache $cache) {
     };
 }, ['pools', 'cache']);
 
-CLI::setResource('queueForStatsUsage', function (Publisher $publisher) {
+CLI::setResource('queueForStatsUsage', function (Connection $publisher) {
     return new StatsUsage($publisher);
 }, ['publisher']);
 CLI::setResource('queueForStatsResources', function (Publisher $publisher) {
     return new StatsResources($publisher);
 }, ['publisher']);
 CLI::setResource('publisher', function (Group $pools) {
-    return new BrokerPool(publisher: $pools->get('publisher'));
+    return $pools->get('publisher')->pop()->getResource();
 }, ['pools']);
 CLI::setResource('queueForFunctions', function (Publisher $publisher) {
     return new Func($publisher);

app/controllers/api/health.php

@@ -3,16 +3,13 @@
 use Appwrite\ClamAV\Network;
 use Appwrite\Event\Event;
 use Appwrite\Extend\Exception;
-use Appwrite\PubSub\Adapter\Pool as PubSubPool;
 use Appwrite\SDK\AuthType;
 use Appwrite\SDK\ContentType;
 use Appwrite\SDK\Method;
 use Appwrite\SDK\Response as SDKResponse;
 use Appwrite\Utopia\Response;
 use Utopia\App;
-use Utopia\Cache\Adapter\Pool as CachePool;
 use Utopia\Config\Config;
-use Utopia\Database\Adapter\Pool as DatabasePool;
 use Utopia\Database\Document;
 use Utopia\Domains\Validator\PublicDomain;
 use Utopia\Pools\Group;
@@ -37,8 +34,8 @@ App::get('/v1/health')
         namespace: 'health',
         group: 'health',
         name: 'get',
-        description: '/docs/references/health/get.md',
         auth: [AuthType::KEY],
+        description: '/docs/references/health/get.md',
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -73,11 +70,11 @@ App::get('/v1/health/db')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'health',
         name: 'getDB',
         description: '/docs/references/health/get-db.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -89,8 +86,8 @@ App::get('/v1/health/db')
     ->inject('response')
     ->inject('pools')
     ->action(function (Response $response, Group $pools) {
+
         $output = [];
-        $failures = [];
 
         $configs = [
             'Console.DB' => Config::getParam('pools-console'),
@@ -100,7 +97,7 @@ App::get('/v1/health/db')
         foreach ($configs as $key => $config) {
             foreach ($config as $database) {
                 try {
-                    $adapter = new DatabasePool($pools->get($database));
+                    $adapter = $pools->get($database)->pop()->getResource();
 
                     $checkStart = \microtime(true);
 
@@ -111,16 +108,16 @@ App::get('/v1/health/db')
                             'ping' => \round((\microtime(true) - $checkStart) / 1000)
                         ]);
                     } else {
-                        $failures[] = $database;
+                        $failure[] = $database;
                     }
-                } catch (\Throwable) {
-                    $failures[] = $database;
+                } catch (\Throwable $th) {
+                    $failure[] = $database;
                 }
             }
         }
 
-        if (!empty($failures)) {
-            throw new Exception(Exception::GENERAL_SERVER_ERROR, 'DB failure on: ' . implode(", ", $failures));
+        if (!empty($failure)) {
+            throw new Exception(Exception::GENERAL_SERVER_ERROR, 'DB failure on: ' . implode(", ", $failure));
         }
 
         $response->dynamic(new Document([
@@ -134,11 +131,11 @@ App::get('/v1/health/cache')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'health',
         name: 'getCache',
         description: '/docs/references/health/get-cache.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -150,39 +147,44 @@ App::get('/v1/health/cache')
     ->inject('response')
     ->inject('pools')
     ->action(function (Response $response, Group $pools) {
+
         $output = [];
-        $failures = [];
 
         $configs = [
             'Cache' => Config::getParam('pools-cache'),
         ];
 
         foreach ($configs as $key => $config) {
-            foreach ($config as $cache) {
+            foreach ($config as $database) {
                 try {
-                    $adapter = new CachePool($pools->get($cache));
+                    /** @var \Utopia\Cache\Adapter $adapter */
+                    $adapter = $pools->get($database)->pop()->getResource();
 
                     $checkStart = \microtime(true);
 
                     if ($adapter->ping()) {
                         $output[] = new Document([
-                            'name' => $key . " ($cache)",
+                            'name' => $key . " ($database)",
                             'status' => 'pass',
                             'ping' => \round((\microtime(true) - $checkStart) / 1000)
                         ]);
                     } else {
-                        $failures[] = $cache;
+                        $output[] = new Document([
+                            'name' => $key . " ($database)",
+                            'status' => 'fail',
+                            'ping' => \round((\microtime(true) - $checkStart) / 1000)
+                        ]);
                     }
-                } catch (\Throwable) {
-                    $failures[] = $cache;
+                } catch (\Throwable $th) {
+                    $output[] = new Document([
+                        'name' => $key . " ($database)",
+                        'status' => 'fail',
+                        'ping' => \round((\microtime(true) - $checkStart) / 1000)
+                    ]);
                 }
             }
         }
 
-        if (!empty($failures)) {
-            throw new Exception(Exception::GENERAL_SERVER_ERROR, 'Cache failure on: ' . implode(", ", $failures));
-        }
-
         $response->dynamic(new Document([
             'statuses' => $output,
             'total' => count($output),
@@ -194,11 +196,11 @@ App::get('/v1/health/pubsub')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'health',
         name: 'getPubSub',
         description: '/docs/references/health/get-pubsub.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -210,39 +212,44 @@ App::get('/v1/health/pubsub')
     ->inject('response')
     ->inject('pools')
     ->action(function (Response $response, Group $pools) {
+
         $output = [];
-        $failures = [];
 
         $configs = [
             'PubSub' => Config::getParam('pools-pubsub'),
         ];
 
         foreach ($configs as $key => $config) {
-            foreach ($config as $pubsub) {
+            foreach ($config as $database) {
                 try {
-                    $adapter = new PubSubPool($pools->get($pubsub));
+                    /** @var \Appwrite\PubSub\Adapter $adapter */
+                    $adapter = $pools->get($database)->pop()->getResource();
 
                     $checkStart = \microtime(true);
 
                     if ($adapter->ping()) {
                         $output[] = new Document([
-                            'name' => $key . " ($pubsub)",
+                            'name' => $key . " ($database)",
                             'status' => 'pass',
                             'ping' => \round((\microtime(true) - $checkStart) / 1000)
                         ]);
                     } else {
-                        $failures[] = $pubsub;
+                        $output[] = new Document([
+                            'name' => $key . " ($database)",
+                            'status' => 'fail',
+                            'ping' => \round((\microtime(true) - $checkStart) / 1000)
+                        ]);
                     }
-                } catch (\Throwable) {
-                    $failures[] = $pubsub;
+                } catch (\Throwable $th) {
+                    $output[] = new Document([
+                        'name' => $key . " ($database)",
+                        'status' => 'fail',
+                        'ping' => \round((\microtime(true) - $checkStart) / 1000)
+                    ]);
                 }
             }
         }
 
-        if (!empty($failures)) {
-            throw new Exception(Exception::GENERAL_SERVER_ERROR, 'Pubsub failure on: ' . implode(", ", $failures));
-        }
-
         $response->dynamic(new Document([
             'statuses' => $output,
             'total' => count($output),
@@ -254,11 +261,11 @@ App::get('/v1/health/time')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'health',
         name: 'getTime',
         description: '/docs/references/health/get-time.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -318,11 +325,11 @@ App::get('/v1/health/queue/webhooks')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'queue',
         name: 'getQueueWebhooks',
         description: '/docs/references/health/get-queue-webhooks.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -344,18 +351,18 @@ App::get('/v1/health/queue/webhooks')
         }
 
         $response->dynamic(new Document([ 'size' => $size ]), Response::MODEL_HEALTH_QUEUE);
-    });
+    }, ['response']);
 
 App::get('/v1/health/queue/logs')
     ->desc('Get logs queue')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'queue',
         name: 'getQueueLogs',
         description: '/docs/references/health/get-queue-logs.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -377,18 +384,18 @@ App::get('/v1/health/queue/logs')
         }
 
         $response->dynamic(new Document([ 'size' => $size ]), Response::MODEL_HEALTH_QUEUE);
-    });
+    }, ['response']);
 
 App::get('/v1/health/certificate')
     ->desc('Get the SSL certificate for a domain')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'health',
         name: 'getCertificate',
         description: '/docs/references/health/get-certificate.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -434,18 +441,18 @@ App::get('/v1/health/certificate')
             'validTo' => $certificatePayload['validTo_time_t'],
             'signatureTypeSN' => $certificatePayload['signatureTypeSN'],
         ]), Response::MODEL_HEALTH_CERTIFICATE);
-    });
+    }, ['response']);
 
 App::get('/v1/health/queue/certificates')
     ->desc('Get certificates queue')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'queue',
         name: 'getQueueCertificates',
         description: '/docs/references/health/get-queue-certificates.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -467,18 +474,18 @@ App::get('/v1/health/queue/certificates')
         }
 
         $response->dynamic(new Document([ 'size' => $size ]), Response::MODEL_HEALTH_QUEUE);
-    });
+    }, ['response']);
 
 App::get('/v1/health/queue/builds')
     ->desc('Get builds queue')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'queue',
         name: 'getQueueBuilds',
         description: '/docs/references/health/get-queue-builds.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -500,18 +507,18 @@ App::get('/v1/health/queue/builds')
         }
 
         $response->dynamic(new Document([ 'size' => $size ]), Response::MODEL_HEALTH_QUEUE);
-    });
+    }, ['response']);
 
 App::get('/v1/health/queue/databases')
     ->desc('Get databases queue')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'queue',
         name: 'getQueueDatabases',
         description: '/docs/references/health/get-queue-databases.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -534,18 +541,18 @@ App::get('/v1/health/queue/databases')
         }
 
         $response->dynamic(new Document([ 'size' => $size ]), Response::MODEL_HEALTH_QUEUE);
-    });
+    }, ['response']);
 
 App::get('/v1/health/queue/deletes')
     ->desc('Get deletes queue')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'queue',
         name: 'getQueueDeletes',
         description: '/docs/references/health/get-queue-deletes.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -567,18 +574,18 @@ App::get('/v1/health/queue/deletes')
         }
 
         $response->dynamic(new Document([ 'size' => $size ]), Response::MODEL_HEALTH_QUEUE);
-    });
+    }, ['response']);
 
 App::get('/v1/health/queue/mails')
     ->desc('Get mails queue')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'queue',
         name: 'getQueueMails',
         description: '/docs/references/health/get-queue-mails.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -600,18 +607,18 @@ App::get('/v1/health/queue/mails')
         }
 
         $response->dynamic(new Document([ 'size' => $size ]), Response::MODEL_HEALTH_QUEUE);
-    });
+    }, ['response']);
 
 App::get('/v1/health/queue/messaging')
     ->desc('Get messaging queue')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'queue',
         name: 'getQueueMessaging',
         description: '/docs/references/health/get-queue-messaging.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -633,18 +640,18 @@ App::get('/v1/health/queue/messaging')
         }
 
         $response->dynamic(new Document([ 'size' => $size ]), Response::MODEL_HEALTH_QUEUE);
-    });
+    }, ['response']);
 
 App::get('/v1/health/queue/migrations')
     ->desc('Get migrations queue')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'queue',
         name: 'getQueueMigrations',
         description: '/docs/references/health/get-queue-migrations.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -666,18 +673,18 @@ App::get('/v1/health/queue/migrations')
         }
 
         $response->dynamic(new Document([ 'size' => $size ]), Response::MODEL_HEALTH_QUEUE);
-    });
+    }, ['response']);
 
 App::get('/v1/health/queue/functions')
     ->desc('Get functions queue')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'queue',
         name: 'getQueueFunctions',
         description: '/docs/references/health/get-queue-functions.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -699,18 +706,18 @@ App::get('/v1/health/queue/functions')
         }
 
         $response->dynamic(new Document([ 'size' => $size ]), Response::MODEL_HEALTH_QUEUE);
-    });
+    }, ['response']);
 
 App::get('/v1/health/queue/stats-resources')
     ->desc('Get stats  resources queue')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'queue',
         name: 'getQueueStatsResources',
         description: '/docs/references/health/get-queue-stats-resources.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -739,11 +746,11 @@ App::get('/v1/health/queue/stats-usage')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'queue',
         name: 'getQueueUsage',
         description: '/docs/references/health/get-queue-stats-usage.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -772,11 +779,11 @@ App::get('/v1/health/storage/local')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'storage',
         name: 'getStorageLocal',
         description: '/docs/references/health/get-storage-local.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -822,11 +829,11 @@ App::get('/v1/health/storage')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'storage',
         name: 'getStorage',
         description: '/docs/references/health/get-storage.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -870,11 +877,11 @@ App::get('/v1/health/anti-virus')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'health',
         name: 'getAntivirus',
         description: '/docs/references/health/get-storage-anti-virus.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,
@@ -916,11 +923,11 @@ App::get('/v1/health/queue/failed/:name')
     ->groups(['api', 'health'])
     ->label('scope', 'health.read')
     ->label('sdk', new Method(
+        auth: [AuthType::KEY],
         namespace: 'health',
         group: 'queue',
         name: 'getFailedJobs',
         description: '/docs/references/health/get-failed-queue-jobs.md',
-        auth: [AuthType::KEY],
         responses: [
             new SDKResponse(
                 code: Response::STATUS_CODE_OK,

app/controllers/api/projects.php

@@ -24,7 +24,6 @@ use Utopia\App;
 use Utopia\Audit\Audit;
 use Utopia\Cache\Cache;
 use Utopia\Config\Config;
-use Utopia\Database\Adapter\Pool as DatabasePool;
 use Utopia\Database\Database;
 use Utopia\Database\DateTime;
 use Utopia\Database\Document;
@@ -224,7 +223,7 @@ App::post('/v1/projects')
         $sharedTables = $sharedTablesV1 || $sharedTablesV2;
 
         if (!$sharedTablesV2) {
-            $adapter = new DatabasePool($pools->get($dsn->getHost()));
+            $adapter = $pools->get($dsn->getHost())->pop()->getResource();
             $dbForProject = new Database($adapter, $cache);
 
             if ($sharedTables) {

app/http.php

@@ -14,11 +14,9 @@ use Utopia\App;
 use Utopia\Audit\Audit;
 use Utopia\CLI\Console;
 use Utopia\Config\Config;
-use Utopia\Database\Adapter\Pool as DatabasePool;
 use Utopia\Database\Database;
 use Utopia\Database\DateTime;
 use Utopia\Database\Document;
-use Utopia\Database\Exception\Duplicate as DuplicateException;
 use Utopia\Database\Helpers\ID;
 use Utopia\Database\Helpers\Permission;
 use Utopia\Database\Helpers\Role;
@@ -169,7 +167,7 @@ function createDatabase(App $app, string $resourceKey, string $dbName, array $co
     $sleep = 1;
     $attempts = 0;
 
-    while (true) {
+    do {
         try {
             $attempts++;
             $resource = $app->getResource($resourceKey);
@@ -178,12 +176,13 @@ function createDatabase(App $app, string $resourceKey, string $dbName, array $co
             break; // exit loop on success
         } catch (\Exception $e) {
             Console::warning("  └── Database not ready. Retrying connection ({$attempts})...");
+            $pools->reclaim();
             if ($attempts >= $max) {
                 throw new \Exception('  └── Failed to connect to database: ' . $e->getMessage());
             }
             sleep($sleep);
         }
-    }
+    } while ($attempts < $max);
 
     Console::success("[Setup] - $dbName database init started...");
 
@@ -319,7 +318,11 @@ $http->on(Constant::EVENT_START, function (Server $http) use ($payloadSize, $reg
         $cache = $app->getResource('cache');
 
         foreach ($sharedTablesV2 as $hostname) {
-            $adapter = new DatabasePool($pools->get($hostname));
+            $adapter = $pools
+                ->get($hostname)
+                ->pop()
+                ->getResource();
+
             $dbForProject = (new Database($adapter, $cache))
                 ->setDatabase('appwrite')
                 ->setSharedTables(true)
@@ -329,7 +332,7 @@ $http->on(Constant::EVENT_START, function (Server $http) use ($payloadSize, $reg
             try {
                 Console::success('[Setup] - Creating project database: ' . $hostname . '...');
                 $dbForProject->create();
-            } catch (DuplicateException) {
+            } catch (Duplicate) {
                 Console::success('[Setup] - Skip: metadata table already exists');
             }
 
@@ -355,6 +358,7 @@ $http->on(Constant::EVENT_START, function (Server $http) use ($payloadSize, $reg
             }
         }
 
+        $pools->reclaim();
         Console::success('[Setup] - Server database init completed...');
     });
 
@@ -469,7 +473,6 @@ $http->on(Constant::EVENT_REQUEST, function (SwooleRequest $swooleRequest, Swool
         Console::error('[Error] Message: ' . $th->getMessage());
         Console::error('[Error] File: ' . $th->getFile());
         Console::error('[Error] Line: ' . $th->getLine());
-        Console::error('[Error] Trace: ' . $th->getTraceAsString());
 
         $swooleResponse->setStatusCode(500);
 
@@ -487,11 +490,13 @@ $http->on(Constant::EVENT_REQUEST, function (SwooleRequest $swooleRequest, Swool
         ];
 
         $swooleResponse->end(\json_encode($output));
+    } finally {
+        $pools->reclaim();
     }
 });
 
 // Fetch domains every `DOMAIN_SYNC_TIMER` seconds and update in the memory
-$http->on(Constant::EVENT_TASK, function () use ($register, $domains) {
+$http->on('Task', function () use ($register, $domains) {
     $lastSyncUpdate = null;
     $pools = $register->get('pools');
     App::setResource('pools', fn () => $pools);

app/init/registers.php

@@ -215,13 +215,13 @@ $register->set('pools', function () {
                 'mysql',
                 'mariadb' => function () use ($dsnHost, $dsnPort, $dsnUser, $dsnPass, $dsnDatabase) {
                     return new PDOProxy(function () use ($dsnHost, $dsnPort, $dsnUser, $dsnPass, $dsnDatabase) {
-                        return new PDO("mysql:host={$dsnHost};port={$dsnPort};dbname={$dsnDatabase};charset=utf8mb4", $dsnUser, $dsnPass, [
+                        return new PDO("mysql:host={$dsnHost};port={$dsnPort};dbname={$dsnDatabase};charset=utf8mb4", $dsnUser, $dsnPass, array(
                             PDO::ATTR_TIMEOUT => 3, // Seconds
                             PDO::ATTR_PERSISTENT => false,
                             PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
                             PDO::ATTR_EMULATE_PREPARES => true,
                             PDO::ATTR_STRINGIFY_FETCHES => true
-                        ]);
+                        ));
                     });
                 },
                 'redis' => function () use ($dsnHost, $dsnPort, $dsnPass) {

app/init/resources.php

@@ -24,12 +24,10 @@ use Appwrite\Utopia\Request;
 use Executor\Executor;
 use Utopia\Abuse\Adapters\TimeLimit\Redis as TimeLimitRedis;
 use Utopia\App;
-use Utopia\Cache\Adapter\Pool as CachePool;
 use Utopia\Cache\Adapter\Sharding;
 use Utopia\Cache\Cache;
 use Utopia\CLI\Console;
 use Utopia\Config\Config;
-use Utopia\Database\Adapter\Pool as DatabasePool;
 use Utopia\Database\Database;
 use Utopia\Database\Document;
 use Utopia\Database\Helpers\ID;
@@ -39,7 +37,6 @@ use Utopia\DSN\DSN;
 use Utopia\Locale\Locale;
 use Utopia\Logger\Log;
 use Utopia\Pools\Group;
-use Utopia\Queue\Broker\Pool as BrokerPool;
 use Utopia\Queue\Publisher;
 use Utopia\Storage\Device;
 use Utopia\Storage\Device\AWS;
@@ -75,10 +72,10 @@ App::setResource('localeCodes', function () {
 
 // Queues
 App::setResource('publisher', function (Group $pools) {
-    return new BrokerPool(publisher: $pools->get('publisher'));
+    return $pools->get('publisher')->pop()->getResource();
 }, ['pools']);
 App::setResource('consumer', function (Group $pools) {
-    return new BrokerPool(consumer: $pools->get('consumer'));
+    return $pools->get('consumer')->pop()->getResource();
 }, ['pools']);
 App::setResource('queueForMessaging', function (Publisher $publisher) {
     return new Messaging($publisher);
@@ -332,8 +329,12 @@ App::setResource('dbForProject', function (Group $pools, Database $dbForPlatform
         $dsn = new DSN('mysql://' . $project->getAttribute('database'));
     }
 
-    $adapter = new DatabasePool($pools->get($dsn->getHost()));
-    $database = new Database($adapter, $cache);
+    $dbAdapter = $pools
+        ->get($dsn->getHost())
+        ->pop()
+        ->getResource();
+
+    $database = new Database($dbAdapter, $cache);
 
     $database
         ->setMetadata('host', \gethostname())
@@ -359,8 +360,12 @@ App::setResource('dbForProject', function (Group $pools, Database $dbForPlatform
 }, ['pools', 'dbForPlatform', 'cache', 'project']);
 
 App::setResource('dbForPlatform', function (Group $pools, Cache $cache) {
-    $adapter = new DatabasePool($pools->get('console'));
-    $database = new Database($adapter, $cache);
+    $dbAdapter = $pools
+        ->get('console')
+        ->pop()
+        ->getResource();
+
+    $database = new Database($dbAdapter, $cache);
 
     $database
         ->setNamespace('_console')
@@ -373,7 +378,7 @@ App::setResource('dbForPlatform', function (Group $pools, Cache $cache) {
 }, ['pools', 'cache']);
 
 App::setResource('getProjectDB', function (Group $pools, Database $dbForPlatform, $cache) {
-    $databases = [];
+    $databases = []; // TODO: @Meldiron This should probably be responsibility of utopia-php/pools
 
     return function (Document $project) use ($pools, $dbForPlatform, $cache, &$databases) {
         if ($project->isEmpty() || $project->getId() === 'console') {
@@ -415,8 +420,12 @@ App::setResource('getProjectDB', function (Group $pools, Database $dbForPlatform
             return $database;
         }
 
-        $adapter = new DatabasePool($pools->get($dsn->getHost()));
-        $database = new Database($adapter, $cache);
+        $dbAdapter = $pools
+            ->get($dsn->getHost())
+            ->pop()
+            ->getResource();
+
+        $database = new Database($dbAdapter, $cache);
         $databases[$dsn->getHost()] = $database;
         $configure($database);
 
@@ -426,15 +435,21 @@ App::setResource('getProjectDB', function (Group $pools, Database $dbForPlatform
 
 App::setResource('getLogsDB', function (Group $pools, Cache $cache) {
     $database = null;
-
-    return function (?Document $project = null) use ($pools, $cache, &$database) {
+    return function (?Document $project = null) use ($pools, $cache, $database) {
         if ($database !== null && $project !== null && !$project->isEmpty() && $project->getId() !== 'console') {
             $database->setTenant($project->getInternalId());
             return $database;
         }
 
-        $adapter = new DatabasePool($pools->get('logs'));
-        $database = new Database($adapter, $cache);
+        $dbAdapter = $pools
+            ->get('logs')
+            ->pop()
+            ->getResource();
+
+        $database = new Database(
+            $dbAdapter,
+            $cache
+        );
 
         $database
             ->setSharedTables(true)
@@ -458,7 +473,10 @@ App::setResource('cache', function (Group $pools, Telemetry $telemetry) {
     $adapters = [];
 
     foreach ($list as $value) {
-        $adapters[] = new CachePool($pools->get($value));
+        $adapters[] = $pools
+            ->get($value)
+            ->pop()
+            ->getResource();
     }
 
     $cache = new Cache(new Sharding($adapters));

app/realtime.php

@@ -5,7 +5,6 @@ use Appwrite\Extend\Exception;
 use Appwrite\Extend\Exception as AppwriteException;
 use Appwrite\Messaging\Adapter\Realtime;
 use Appwrite\Network\Validator\Origin;
-use Appwrite\PubSub\Adapter\Pool as PubSubPool;
 use Appwrite\Utopia\Request;
 use Appwrite\Utopia\Response;
 use Swoole\Http\Request as SwooleRequest;
@@ -16,12 +15,10 @@ use Swoole\Timer;
 use Utopia\Abuse\Abuse;
 use Utopia\Abuse\Adapters\TimeLimit\Redis as TimeLimitRedis;
 use Utopia\App;
-use Utopia\Cache\Adapter\Pool as CachePool;
 use Utopia\Cache\Adapter\Sharding;
 use Utopia\Cache\Cache;
 use Utopia\CLI\Console;
 use Utopia\Config\Config;
-use Utopia\Database\Adapter\Pool as DatabasePool;
 use Utopia\Database\Database;
 use Utopia\Database\DateTime;
 use Utopia\Database\Document;
@@ -31,15 +28,13 @@ use Utopia\Database\Query;
 use Utopia\Database\Validator\Authorization;
 use Utopia\DSN\DSN;
 use Utopia\Logger\Log;
-use Utopia\Pools\Group;
-use Utopia\Registry\Registry;
 use Utopia\System\System;
 use Utopia\Telemetry\Adapter\None as NoTelemetry;
 use Utopia\WebSocket\Adapter;
 use Utopia\WebSocket\Server;
 
 /**
- * @var Registry $register
+ * @var \Utopia\Registry\Registry $register
  */
 require_once __DIR__ . '/init.php';
 
@@ -51,17 +46,17 @@ if (!function_exists('getConsoleDB')) {
     {
         global $register;
 
-        static $database = null;
-
-        if ($database !== null) {
-            return $database;
-        }
-
-        /** @var Group $pools */
+        /** @var \Utopia\Pools\Group $pools */
         $pools = $register->get('pools');
 
-        $adapter = new DatabasePool($pools->get('console'));
-        $database = new Database($adapter, getCache());
+        $dbAdapter = $pools
+            ->get('console')
+            ->pop()
+            ->getResource()
+        ;
+
+        $database = new Database($dbAdapter, getCache());
+
         $database
             ->setNamespace('_console')
             ->setMetadata('host', \gethostname())
@@ -77,13 +72,7 @@ if (!function_exists('getProjectDB')) {
     {
         global $register;
 
-        static $databases = [];
-
-        if (isset($databases[$project->getInternalId()])) {
-            return $databases[$project->getInternalId()];
-        }
-
-        /** @var Group $pools */
+        /** @var \Utopia\Pools\Group $pools */
         $pools = $register->get('pools');
 
         if ($project->isEmpty() || $project->getId() === 'console') {
@@ -97,7 +86,11 @@ if (!function_exists('getProjectDB')) {
             $dsn = new DSN('mysql://' . $project->getAttribute('database'));
         }
 
-        $adapter = new DatabasePool($pools->get($dsn->getHost()));
+        $adapter = $pools
+            ->get($dsn->getHost())
+            ->pop()
+            ->getResource();
+
         $database = new Database($adapter, getCache());
 
         $sharedTables = \explode(',', System::getEnv('_APP_DATABASE_SHARED_TABLES', ''));
@@ -118,7 +111,7 @@ if (!function_exists('getProjectDB')) {
             ->setMetadata('host', \gethostname())
             ->setMetadata('project', $project->getId());
 
-        return $databases[$project->getInternalId()] = $database;
+        return $database;
     }
 }
 
@@ -128,22 +121,20 @@ if (!function_exists('getCache')) {
     {
         global $register;
 
-        static $cache = null;
-
-        if ($cache !== null) {
-            return $cache;
-        }
-
-        $pools = $register->get('pools'); /** @var Group $pools */
+        $pools = $register->get('pools'); /** @var \Utopia\Pools\Group $pools */
 
         $list = Config::getParam('pools-cache', []);
         $adapters = [];
 
         foreach ($list as $value) {
-            $adapters[] = new CachePool($pools->get($value));
+            $adapters[] = $pools
+                ->get($value)
+                ->pop()
+                ->getResource()
+            ;
         }
 
-        return $cache = new Cache(new Sharding($adapters));
+        return new Cache(new Sharding($adapters));
     }
 }
 
@@ -151,12 +142,6 @@ if (!function_exists('getCache')) {
 if (!function_exists('getRedis')) {
     function getRedis(): \Redis
     {
-        static $redis = null;
-
-        if ($redis !== null) {
-            return $redis;
-        }
-
         $host = System::getEnv('_APP_REDIS_HOST', 'localhost');
         $port = System::getEnv('_APP_REDIS_PORT', 6379);
         $pass = System::getEnv('_APP_REDIS_PASS', '');
@@ -175,39 +160,21 @@ if (!function_exists('getRedis')) {
 if (!function_exists('getTimelimit')) {
     function getTimelimit(): TimeLimitRedis
     {
-        static $timelimit = null;
-
-        if ($timelimit !== null) {
-            return $timelimit;
-        }
-
-        return $timelimit = new TimeLimitRedis("", 0, 1, getRedis());
+        return new TimeLimitRedis("", 0, 1, getRedis());
     }
 }
 
 if (!function_exists('getRealtime')) {
     function getRealtime(): Realtime
     {
-        static $realtime = null;
-
-        if ($realtime !== null) {
-            return $realtime;
-        }
-
-        return $realtime = new Realtime();
+        return new Realtime();
     }
 }
 
 if (!function_exists('getTelemetry')) {
     function getTelemetry(int $workerId): Utopia\Telemetry\Adapter
     {
-        static $telemetry = null;
-
-        if ($telemetry !== null) {
-            return $telemetry;
-        }
-
-        return $telemetry = new NoTelemetry();
+        return new NoTelemetry();
     }
 }
 
@@ -306,6 +273,7 @@ $server->onStart(function () use ($stats, $register, $containerId, &$statsDocume
                 sleep(DATABASE_RECONNECT_SLEEP);
             }
         } while (true);
+        $register->get('pools')->reclaim();
     });
 
     /**
@@ -331,7 +299,9 @@ $server->onStart(function () use ($stats, $register, $containerId, &$statsDocume
 
                 Authorization::skip(fn () => $database->updateDocument('realtime', $statsDocument->getId(), $statsDocument));
             } catch (Throwable $th) {
-                $logError($th, "updateWorkerDocument");
+                call_user_func($logError, $th, "updateWorkerDocument");
+            } finally {
+                $register->get('pools')->reclaim();
             }
         });
     }
@@ -400,6 +370,8 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats,
                         'data' => $event['data']
                     ]));
                 }
+
+                $register->get('pools')->reclaim();
             }
         }
         /**
@@ -435,8 +407,8 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats,
             }
             $start = time();
 
-            $pubsub = new PubSubPool($register->get('pools')->get('pubsub'));
-
+            /** @var \Appwrite\PubSub\Adapter $pubsub */
+            $pubsub = $register->get('pools')->get('pubsub')->pop()->getResource();
             if ($pubsub->ping(true)) {
                 $attempts = 0;
                 Console::success('Pub/sub connection established (worker: ' . $workerId . ')');
@@ -464,6 +436,8 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats,
 
                         $realtime->unsubscribe($connection);
                         $realtime->subscribe($projectId, $connection, $roles, $channels);
+
+                        $register->get('pools')->reclaim();
                     }
                 }
 
@@ -489,12 +463,14 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats,
                 }
             });
         } catch (Throwable $th) {
-            $logError($th, "pubSubConnection");
+            call_user_func($logError, $th, "pubSubConnection");
 
             Console::error('Pub/sub error: ' . $th->getMessage());
             $attempts++;
             sleep(DATABASE_RECONNECT_SLEEP);
             continue;
+        } finally {
+            $register->get('pools')->reclaim();
         }
     }
 
@@ -596,7 +572,7 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server,
         $stats->incr($project->getId(), 'connections');
         $stats->incr($project->getId(), 'connectionsTotal');
     } catch (Throwable $th) {
-        $logError($th, "initServer");
+        call_user_func($logError, $th, "initServer");
 
         // Handle SQL error code is 'HY000'
         $code = $th->getCode();
@@ -620,6 +596,8 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server,
             Console::error('[Error] Code: ' . $response['data']['code']);
             Console::error('[Error] Message: ' . $response['data']['message']);
         }
+    } finally {
+        $register->get('pools')->reclaim();
     }
 });
 
@@ -718,6 +696,8 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re
         if ($th->getCode() === 1008) {
             $server->close($connection, $th->getCode());
         }
+    } finally {
+        $register->get('pools')->reclaim();
     }
 });
 

app/worker.php

@@ -20,12 +20,10 @@ use Appwrite\Platform\Appwrite;
 use Executor\Executor;
 use Swoole\Runtime;
 use Utopia\Abuse\Adapters\TimeLimit\Redis as TimeLimitRedis;
-use Utopia\Cache\Adapter\Pool as CachePool;
 use Utopia\Cache\Adapter\Sharding;
 use Utopia\Cache\Cache;
 use Utopia\CLI\Console;
 use Utopia\Config\Config;
-use Utopia\Database\Adapter\Pool as DatabasePool;
 use Utopia\Database\Database;
 use Utopia\Database\DateTime;
 use Utopia\Database\Document;
@@ -35,7 +33,6 @@ use Utopia\Logger\Log;
 use Utopia\Logger\Logger;
 use Utopia\Platform\Service;
 use Utopia\Pools\Group;
-use Utopia\Queue\Broker\Pool as BrokerPool;
 use Utopia\Queue\Message;
 use Utopia\Queue\Publisher;
 use Utopia\Queue\Server;
@@ -43,17 +40,21 @@ use Utopia\Registry\Registry;
 use Utopia\System\System;
 
 Authorization::disable();
-Runtime::enableCoroutine();
+Runtime::enableCoroutine(SWOOLE_HOOK_ALL);
 
 Server::setResource('register', fn () => $register);
 
 Server::setResource('dbForPlatform', function (Cache $cache, Registry $register) {
     $pools = $register->get('pools');
-    $adapter = new DatabasePool($pools->get('console'));
-    $dbForPlatform = new Database($adapter, $cache);
-    $dbForPlatform->setNamespace('_console');
+    $database = $pools
+        ->get('console')
+        ->pop()
+        ->getResource();
 
-    return $dbForPlatform;
+    $adapter = new Database($database, $cache);
+    $adapter->setNamespace('_console');
+
+    return $adapter;
 }, ['cache', 'register']);
 
 Server::setResource('project', function (Message $message, Database $dbForPlatform) {
@@ -81,9 +82,20 @@ Server::setResource('dbForProject', function (Cache $cache, Registry $register,
         $dsn = new DSN('mysql://' . $project->getAttribute('database'));
     }
 
-    $adapter = new DatabasePool($pools->get($dsn->getHost()));
+    $adapter = $pools
+        ->get($dsn->getHost())
+        ->pop()
+        ->getResource();
+
     $database = new Database($adapter, $cache);
 
+    try {
+        $dsn = new DSN($project->getAttribute('database'));
+    } catch (\InvalidArgumentException) {
+        // TODO: Temporary until all projects are using shared tables
+        $dsn = new DSN('mysql://' . $project->getAttribute('database'));
+    }
+
     $sharedTables = \explode(',', System::getEnv('_APP_DATABASE_SHARED_TABLES', ''));
 
     if (\in_array($dsn->getHost(), $sharedTables)) {
@@ -138,8 +150,12 @@ Server::setResource('getProjectDB', function (Group $pools, Database $dbForPlatf
             return $database;
         }
 
-        $adapter = new DatabasePool($pools->get($dsn->getHost()));
-        $database = new Database($adapter, $cache);
+        $dbAdapter = $pools
+            ->get($dsn->getHost())
+            ->pop()
+            ->getResource();
+
+        $database = new Database($dbAdapter, $cache);
 
         $databases[$dsn->getHost()] = $database;
 
@@ -171,8 +187,15 @@ Server::setResource('getLogsDB', function (Group $pools, Cache $cache) {
             return $database;
         }
 
-        $adapter = new DatabasePool($pools->get('logs'));
-        $database = new Database($adapter, $cache);
+        $dbAdapter = $pools
+            ->get('logs')
+            ->pop()
+            ->getResource();
+
+        $database = new Database(
+            $dbAdapter,
+            $cache
+        );
 
         $database
             ->setSharedTables(true)
@@ -210,7 +233,11 @@ Server::setResource('cache', function (Registry $register) {
     $adapters = [];
 
     foreach ($list as $value) {
-        $adapters[] = new CachePool($pools->get($value));
+        $adapters[] = $pools
+            ->get($value)
+            ->pop()
+            ->getResource()
+        ;
     }
 
     return new Cache(new Sharding($adapters));
@@ -240,11 +267,11 @@ Server::setResource('timelimit', function (\Redis $redis) {
 Server::setResource('log', fn () => new Log());
 
 Server::setResource('publisher', function (Group $pools) {
-    return new BrokerPool(publisher: $pools->get('publisher'));
+    return $pools->get('publisher')->pop()->getResource();
 }, ['pools']);
 
 Server::setResource('consumer', function (Group $pools) {
-    return new BrokerPool(consumer: $pools->get('consumer'));
+    return $pools->get('consumer')->pop()->getResource();
 }, ['pools']);
 
 Server::setResource('queueForStatsUsage', function (Publisher $publisher) {
@@ -421,6 +448,13 @@ try {
 
 $worker = $platform->getWorker();
 
+$worker
+    ->shutdown()
+    ->inject('pools')
+    ->action(function (Group $pools) {
+        $pools->reclaim();
+    });
+
 $worker
     ->error()
     ->inject('error')
@@ -428,7 +462,8 @@ $worker
     ->inject('log')
     ->inject('pools')
     ->inject('project')
-    ->action(function (Throwable $error, ?Logger $logger, Log $log, Group $pools, Document $project) use ($worker, $queueName) {
+    ->action(function (Throwable $error, ?Logger $logger, Log $log, Group $pools, Document $project) use ($queueName) {
+        $pools->reclaim();
         $version = System::getEnv('_APP_VERSION', 'UNKNOWN');
 
         if ($logger) {

composer.json

@@ -48,7 +48,7 @@
         "utopia-php/abuse": "0.52.*",
         "utopia-php/analytics": "0.10.*",
         "utopia-php/audit": "0.55.*",
-        "utopia-php/cache": "0.13.*",
+        "utopia-php/cache": "0.12.*",
         "utopia-php/cli": "0.15.*",
         "utopia-php/config": "0.2.*",
         "utopia-php/database": "0.66.*",
@@ -65,7 +65,7 @@
         "utopia-php/platform": "0.7.*",
         "utopia-php/pools": "0.8.*",
         "utopia-php/preloader": "0.2.*",
-        "utopia-php/queue": "0.10.*",
+        "utopia-php/queue": "0.9.*",
         "utopia-php/registry": "0.5.*",
         "utopia-php/storage": "0.18.*",
         "utopia-php/swoole": "0.8.*",

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": "43dc7ad0d25793a58604b9d4d40abff4",
+    "content-hash": "f8a8967327763c5b85721583bd78c1de",
     "packages": [
         {
             "name": "adhocore/jwt",
@@ -3300,16 +3300,16 @@
         },
         {
             "name": "utopia-php/cache",
-            "version": "0.13.0",
+            "version": "0.12.0",
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/cache.git",
-                "reference": "dee01dec33a211644d60f6cfa56b1b8176d3fae3"
+                "reference": "646038f1d470b759c129348be8fc14da3c00bbd9"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/cache/zipball/dee01dec33a211644d60f6cfa56b1b8176d3fae3",
-                "reference": "dee01dec33a211644d60f6cfa56b1b8176d3fae3",
+                "url": "https://api.github.com/repos/utopia-php/cache/zipball/646038f1d470b759c129348be8fc14da3c00bbd9",
+                "reference": "646038f1d470b759c129348be8fc14da3c00bbd9",
                 "shasum": ""
             },
             "require": {
@@ -3317,7 +3317,6 @@
                 "ext-memcached": "*",
                 "ext-redis": "*",
                 "php": ">=8.0",
-                "utopia-php/pools": "0.8.*",
                 "utopia-php/telemetry": "0.1.*"
             },
             "require-dev": {
@@ -3346,9 +3345,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/cache/issues",
-                "source": "https://github.com/utopia-php/cache/tree/0.13.0"
+                "source": "https://github.com/utopia-php/cache/tree/0.12.0"
             },
-            "time": "2025-04-17T04:20:26+00:00"
+            "time": "2025-02-25T09:09:21+00:00"
         },
         {
             "name": "utopia-php/cli",
@@ -3498,23 +3497,23 @@
         },
         {
             "name": "utopia-php/database",
-            "version": "0.66.2",
+            "version": "0.66.0",
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/database.git",
-                "reference": "944d1979c4ba572c746cad957fe34aead338ee34"
+                "reference": "67d2ab418efba31dc76b3564cf043e2b3f98d027"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/database/zipball/944d1979c4ba572c746cad957fe34aead338ee34",
-                "reference": "944d1979c4ba572c746cad957fe34aead338ee34",
+                "url": "https://api.github.com/repos/utopia-php/database/zipball/67d2ab418efba31dc76b3564cf043e2b3f98d027",
+                "reference": "67d2ab418efba31dc76b3564cf043e2b3f98d027",
                 "shasum": ""
             },
             "require": {
                 "ext-mbstring": "*",
                 "ext-pdo": "*",
                 "php": ">=8.1",
-                "utopia-php/cache": "0.13.*",
+                "utopia-php/cache": "0.12.*",
                 "utopia-php/framework": "0.33.*",
                 "utopia-php/pools": "0.8.*"
             },
@@ -3548,9 +3547,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/database/issues",
-                "source": "https://github.com/utopia-php/database/tree/0.66.2"
+                "source": "https://github.com/utopia-php/database/tree/0.66.0"
             },
-            "time": "2025-04-29T23:35:39+00:00"
+            "time": "2025-04-16T07:10:27+00:00"
         },
         {
             "name": "utopia-php/domains",
@@ -4058,16 +4057,16 @@
         },
         {
             "name": "utopia-php/platform",
-            "version": "0.7.5",
+            "version": "0.7.4",
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/platform.git",
-                "reference": "8febd7b6e0c0f2cbd2f4289447bcca97496e4aaf"
+                "reference": "a5b93d8177702ec458c3af9137663133c012b71b"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/platform/zipball/8febd7b6e0c0f2cbd2f4289447bcca97496e4aaf",
-                "reference": "8febd7b6e0c0f2cbd2f4289447bcca97496e4aaf",
+                "url": "https://api.github.com/repos/utopia-php/platform/zipball/a5b93d8177702ec458c3af9137663133c012b71b",
+                "reference": "a5b93d8177702ec458c3af9137663133c012b71b",
                 "shasum": ""
             },
             "require": {
@@ -4076,11 +4075,11 @@
                 "php": ">=8.0",
                 "utopia-php/cli": "0.15.*",
                 "utopia-php/framework": "0.33.*",
-                "utopia-php/queue": "0.10.*"
+                "utopia-php/queue": "0.9.*"
             },
             "require-dev": {
-                "laravel/pint": "1.*",
-                "phpunit/phpunit": "9.*"
+                "laravel/pint": "1.2.*",
+                "phpunit/phpunit": "^9.3"
             },
             "type": "library",
             "autoload": {
@@ -4102,9 +4101,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/platform/issues",
-                "source": "https://github.com/utopia-php/platform/tree/0.7.5"
+                "source": "https://github.com/utopia-php/platform/tree/0.7.4"
             },
-            "time": "2025-04-17T12:20:16+00:00"
+            "time": "2025-03-13T13:00:12+00:00"
         },
         {
             "name": "utopia-php/pools",
@@ -4213,16 +4212,16 @@
         },
         {
             "name": "utopia-php/queue",
-            "version": "0.10.0",
+            "version": "0.9.1",
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/queue.git",
-                "reference": "0eccc559168ea72241c39a4c482d868314666be1"
+                "reference": "32b6f84c55aae761db5a5ae76cc91ca8dbc8bc32"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/queue/zipball/0eccc559168ea72241c39a4c482d868314666be1",
-                "reference": "0eccc559168ea72241c39a4c482d868314666be1",
+                "url": "https://api.github.com/repos/utopia-php/queue/zipball/32b6f84c55aae761db5a5ae76cc91ca8dbc8bc32",
+                "reference": "32b6f84c55aae761db5a5ae76cc91ca8dbc8bc32",
                 "shasum": ""
             },
             "require": {
@@ -4231,7 +4230,6 @@
                 "utopia-php/cli": "0.15.*",
                 "utopia-php/fetch": "0.4.*",
                 "utopia-php/framework": "0.33.*",
-                "utopia-php/pools": "0.8.*",
                 "utopia-php/telemetry": "0.1.*"
             },
             "require-dev": {
@@ -4273,9 +4271,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/queue/issues",
-                "source": "https://github.com/utopia-php/queue/tree/0.10.0"
+                "source": "https://github.com/utopia-php/queue/tree/0.9.1"
             },
-            "time": "2025-04-17T12:15:52+00:00"
+            "time": "2025-03-28T19:49:36+00:00"
         },
         {
             "name": "utopia-php/registry",
@@ -4545,28 +4543,28 @@
         },
         {
             "name": "utopia-php/vcs",
-            "version": "0.9.5",
+            "version": "0.9.4",
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/vcs.git",
-                "reference": "055956545ca7ab4e8688df5de1df3e2833859793"
+                "reference": "1a8d280b176acc99ea8d9e7364b8767cbb206b4a"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/vcs/zipball/055956545ca7ab4e8688df5de1df3e2833859793",
-                "reference": "055956545ca7ab4e8688df5de1df3e2833859793",
+                "url": "https://api.github.com/repos/utopia-php/vcs/zipball/1a8d280b176acc99ea8d9e7364b8767cbb206b4a",
+                "reference": "1a8d280b176acc99ea8d9e7364b8767cbb206b4a",
                 "shasum": ""
             },
             "require": {
                 "adhocore/jwt": "^1.1",
                 "php": ">=8.0",
-                "utopia-php/cache": "0.13.*",
+                "utopia-php/cache": "0.12.*",
                 "utopia-php/framework": "0.*.*",
                 "utopia-php/system": "0.9.*"
             },
             "require-dev": {
-                "laravel/pint": "1.*.*",
-                "phpstan/phpstan": "1.*.*",
+                "laravel/pint": "1.2.*",
+                "phpstan/phpstan": "1.8.*",
                 "phpunit/phpunit": "^9.4"
             },
             "type": "library",
@@ -4589,9 +4587,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/vcs/issues",
-                "source": "https://github.com/utopia-php/vcs/tree/0.9.5"
+                "source": "https://github.com/utopia-php/vcs/tree/0.9.4"
             },
-            "time": "2025-04-17T04:38:49+00:00"
+            "time": "2025-03-13T10:09:45+00:00"
         },
         {
             "name": "utopia-php/websocket",
@@ -5233,16 +5231,16 @@
         },
         {
             "name": "myclabs/deep-copy",
-            "version": "1.13.1",
+            "version": "1.13.0",
             "source": {
                 "type": "git",
                 "url": "https://github.com/myclabs/DeepCopy.git",
-                "reference": "1720ddd719e16cf0db4eb1c6eca108031636d46c"
+                "reference": "024473a478be9df5fdaca2c793f2232fe788e414"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/myclabs/DeepCopy/zipball/1720ddd719e16cf0db4eb1c6eca108031636d46c",
-                "reference": "1720ddd719e16cf0db4eb1c6eca108031636d46c",
+                "url": "https://api.github.com/repos/myclabs/DeepCopy/zipball/024473a478be9df5fdaca2c793f2232fe788e414",
+                "reference": "024473a478be9df5fdaca2c793f2232fe788e414",
                 "shasum": ""
             },
             "require": {
@@ -5281,7 +5279,7 @@
             ],
             "support": {
                 "issues": "https://github.com/myclabs/DeepCopy/issues",
-                "source": "https://github.com/myclabs/DeepCopy/tree/1.13.1"
+                "source": "https://github.com/myclabs/DeepCopy/tree/1.13.0"
             },
             "funding": [
                 {
@@ -5289,7 +5287,7 @@
                     "type": "tidelift"
                 }
             ],
-            "time": "2025-04-29T12:36:36+00:00"
+            "time": "2025-02-12T12:17:51+00:00"
         },
         {
             "name": "nikic/php-parser",

src/Appwrite/Event/Event.php

@@ -286,6 +286,13 @@ class Event
         return $this;
     }
 
+    public function setParamSensitive(string $key): self
+    {
+        $this->sensitive[$key] = true;
+
+        return $this;
+    }
+
     /**
      * Get param of event.
      *

src/Appwrite/Messaging/Adapter/Realtime.php

@@ -2,14 +2,14 @@
 
 namespace Appwrite\Messaging\Adapter;
 
-use Appwrite\Messaging\Adapter as MessagingAdapter;
-use Appwrite\PubSub\Adapter\Pool as PubSubPool;
+use Appwrite\Messaging\Adapter;
 use Utopia\Database\DateTime;
 use Utopia\Database\Document;
 use Utopia\Database\Helpers\ID;
 use Utopia\Database\Helpers\Role;
+use Utopia\Pools\Pool;
 
-class Realtime extends MessagingAdapter
+class Realtime extends Adapter
 {
     /**
      * Connection Tree
@@ -36,12 +36,12 @@ class Realtime extends MessagingAdapter
      */
     public array $subscriptions = [];
 
-    private PubSubPool $redis;
+    private Pool $pubsubPool;
 
     public function __construct()
     {
         global $register;
-        $this->redis = new PubSubPool($register->get('pools')->get('pubsub'));
+        $this->pubsubPool = $register->get('pools')->get('pubsub');
     }
 
     /**
@@ -132,12 +132,11 @@ class Realtime extends MessagingAdapter
      * Sends an event to the Realtime Server
      * @param string $projectId
      * @param array $payload
-     * @param array $events
+     * @param string $event
      * @param array $channels
      * @param array $roles
      * @param array $options
      * @return void
-     * @throws \Exception
      */
     public function send(string $projectId, array $payload, array $events, array $channels, array $roles, array $options = []): void
     {
@@ -148,7 +147,7 @@ class Realtime extends MessagingAdapter
         $permissionsChanged = array_key_exists('permissionsChanged', $options) && $options['permissionsChanged'];
         $userId = array_key_exists('userId', $options) ? $options['userId'] : null;
 
-        $this->redis->publish('realtime', json_encode([
+        $message = [
             'project' => $projectId,
             'roles' => $roles,
             'permissionsChanged' => $permissionsChanged,
@@ -159,7 +158,9 @@ class Realtime extends MessagingAdapter
                 'timestamp' => DateTime::formatTz(DateTime::now()),
                 'payload' => $payload
             ]
-        ]));
+        ];
+
+        $this->pubsubPool->use(fn (\Appwrite\PubSub\Adapter $pubsub) => $pubsub->publish('realtime', json_encode($message)));
     }
 
     /**
@@ -174,9 +175,8 @@ class Realtime extends MessagingAdapter
      *  - 1,121.328 ms (±0.84%) | 1,000,000 Connections / 10,000,000 Subscriptions
      *
      * @param array $event
-     * @return int[]|string[]
      */
-    public function getSubscribers(array $event): array
+    public function getSubscribers(array $event)
     {
 
         $receivers = [];
@@ -230,7 +230,7 @@ class Realtime extends MessagingAdapter
 
         foreach ($channels as $key => $value) {
             switch (true) {
-                case \str_starts_with($key, 'account.'):
+                case strpos($key, 'account.') === 0:
                     unset($channels[$key]);
                     break;
 
@@ -272,7 +272,6 @@ class Realtime extends MessagingAdapter
                 $channels[] = 'account.' . $parts[1];
                 $roles = [Role::user(ID::custom($parts[1]))->toString()];
                 break;
-            case 'migrations':
             case 'rules':
                 $channels[] = 'console';
                 $channels[] = 'projects.' . $project->getId();
@@ -353,6 +352,12 @@ class Realtime extends MessagingAdapter
                     $roles = [Role::team($project->getAttribute('teamId'))->toString()];
                 }
 
+                break;
+            case 'migrations':
+                $channels[] = 'console';
+                $channels[] = 'projects.' . $project->getId();
+                $projectId = 'console';
+                $roles = [Role::team($project->getAttribute('teamId'))->toString()];
                 break;
         }
 

src/Appwrite/Platform/Tasks/Doctor.php

@@ -3,19 +3,14 @@
 namespace Appwrite\Platform\Tasks;
 
 use Appwrite\ClamAV\Network;
-use Appwrite\PubSub\Adapter\Pool as PubSubPool;
-use PHPMailer\PHPMailer\PHPMailer;
+use Appwrite\PubSub\Adapter;
 use Utopia\App;
-use Utopia\Cache\Adapter\Pool as CachePool;
 use Utopia\CLI\Console;
 use Utopia\Config\Config;
-use Utopia\Database\Adapter\Pool as DatabasePool;
 use Utopia\Domains\Domain;
 use Utopia\DSN\DSN;
 use Utopia\Logger\Logger;
 use Utopia\Platform\Action;
-use Utopia\Pools\Group;
-use Utopia\Queue\Broker\Pool as BrokerPool;
 use Utopia\Registry\Registry;
 use Utopia\Storage\Device\Local;
 use Utopia\Storage\Storage;
@@ -81,9 +76,9 @@ class Doctor extends Action
             Console::log('🟢 Abuse protection is enabled');
         }
 
-        $authWhitelistRoot = System::getEnv('_APP_CONSOLE_WHITELIST_ROOT');
-        $authWhitelistEmails = System::getEnv('_APP_CONSOLE_WHITELIST_EMAILS');
-        $authWhitelistIPs = System::getEnv('_APP_CONSOLE_WHITELIST_IPS');
+        $authWhitelistRoot = System::getEnv('_APP_CONSOLE_WHITELIST_ROOT', null);
+        $authWhitelistEmails = System::getEnv('_APP_CONSOLE_WHITELIST_EMAILS', null);
+        $authWhitelistIPs = System::getEnv('_APP_CONSOLE_WHITELIST_IPS', null);
 
         if (
             empty($authWhitelistRoot)
@@ -119,16 +114,19 @@ class Doctor extends Action
             } else {
                 Console::log('🟢 Logging adapter is enabled (' . $providerName . ')');
             }
-        } catch (\Throwable) {
+        } catch (\Throwable $th) {
             Console::log('🔴 Logging adapter is misconfigured');
         }
 
         \usleep(200 * 1000); // Sleep for 0.2 seconds
 
-        Console::log("\n" . '[Connectivity]');
+        try {
+            Console::log("\n" . '[Connectivity]');
+        } catch (\Throwable $th) {
+            //throw $th;
+        }
 
-        /** @var Group $pools */
-        $pools = $register->get('pools');
+        $pools = $register->get('pools'); /** @var \Utopia\Pools\Group $pools */
 
         $configs = [
             'Console.DB' => Config::getParam('pools-console'),
@@ -138,22 +136,20 @@ class Doctor extends Action
         foreach ($configs as $key => $config) {
             foreach ($config as $database) {
                 try {
-                    $adapter = new DatabasePool($pools->get($database));
+                    $adapter = $pools->get($database)->pop()->getResource();
 
                     if ($adapter->ping()) {
                         Console::success('🟢 ' . str_pad("{$key}({$database})", 50, '.') . 'connected');
                     } else {
                         Console::error('🔴 ' . str_pad("{$key}({$database})", 47, '.') . 'disconnected');
                     }
-                } catch (\Throwable) {
+                } catch (\Throwable $th) {
                     Console::error('🔴 ' . str_pad("{$key}.({$database})", 47, '.') . 'disconnected');
                 }
             }
         }
 
-        /** @var Group $pools */
-        $pools = $register->get('pools');
-
+        $pools = $register->get('pools'); /** @var \Utopia\Pools\Group $pools */
         $configs = [
             'Cache' => Config::getParam('pools-cache'),
             'Queue' => Config::getParam('pools-queue'),
@@ -163,18 +159,15 @@ class Doctor extends Action
         foreach ($configs as $key => $config) {
             foreach ($config as $pool) {
                 try {
-                    $adapter = match($key) {
-                        'Cache' => new CachePool($pools->get($pool)),
-                        'Queue' => new BrokerPool($pools->get($pool)),
-                        'PubSub' => new PubSubPool($pools->get($pool)),
-                    };
+                    /** @var Adapter $adapter */
+                    $adapter = $pools->get($pool)->pop()->getResource();
 
                     if ($adapter->ping()) {
                         Console::success('🟢 ' . str_pad("{$key}({$pool})", 50, '.') . 'connected');
                     } else {
                         Console::error('🔴 ' . str_pad("{$key}({$pool})", 47, '.') . 'disconnected');
                     }
-                } catch (\Throwable) {
+                } catch (\Throwable $th) {
                     Console::error('🔴 ' . str_pad("{$key}({$pool})", 47, '.') . 'disconnected');
                 }
             }
@@ -192,14 +185,13 @@ class Doctor extends Action
                 } else {
                     Console::error('🔴 ' . str_pad("Antivirus", 47, '.') . 'disconnected');
                 }
-            } catch (\Throwable) {
+            } catch (\Throwable $th) {
                 Console::error('🔴 ' . str_pad("Antivirus", 47, '.') . 'disconnected');
             }
         }
 
         try {
-            /* @var PHPMailer $mail */
-            $mail = $register->get('smtp');
+            $mail = $register->get('smtp'); /* @var $mail \PHPMailer\PHPMailer\PHPMailer */
 
             $mail->addAddress('demo@example.com', 'Example.com');
             $mail->Subject = 'Test SMTP Connection';
@@ -208,7 +200,7 @@ class Doctor extends Action
 
             $mail->send();
             Console::success('🟢 ' . str_pad("SMTP", 50, '.') . 'connected');
-        } catch (\Throwable) {
+        } catch (\Throwable $th) {
             Console::error('🔴 ' . str_pad("SMTP", 47, '.') . 'disconnected');
         }
 
@@ -282,7 +274,7 @@ class Doctor extends Action
                     Console::error('Failed to check for a newer version' . "\n");
                 }
             }
-        } catch (\Throwable) {
+        } catch (\Throwable $th) {
             Console::error('Failed to check for a newer version' . "\n");
         }
     }

src/Appwrite/Platform/Tasks/ScheduleBase.php

@@ -12,7 +12,6 @@ use Utopia\Database\Query;
 use Utopia\Database\Validator\Authorization;
 use Utopia\Platform\Action;
 use Utopia\Pools\Group;
-use Utopia\Queue\Broker\Pool as BrokerPool;
 use Utopia\System\System;
 
 use function Swoole\Coroutine\run;
@@ -24,8 +23,6 @@ abstract class ScheduleBase extends Action
 
     protected array $schedules = [];
 
-    protected BrokerPool $publisher;
-
     abstract public static function getName(): string;
     abstract public static function getSupportedResource(): string;
     abstract public static function getCollectionId(): string;
@@ -64,8 +61,6 @@ abstract class ScheduleBase extends Action
         Console::title(\ucfirst(static::getSupportedResource()) . ' scheduler V1');
         Console::success(APP_NAME . ' ' . \ucfirst(static::getSupportedResource()) . ' scheduler v1 has started');
 
-        $this->publisher = new BrokerPool($pools->get('publisher'));
-
         /**
          * Extract only necessary attributes to lower memory used.
          *
@@ -137,6 +132,8 @@ abstract class ScheduleBase extends Action
             $latestDocument = \end($results);
         }
 
+        $pools->reclaim();
+
         Console::success("{$total} resources were loaded in " . (\microtime(true) - $loadStart) . " seconds");
 
         Console::success("Starting timers at " . DateTime::now());
@@ -201,6 +198,8 @@ abstract class ScheduleBase extends Action
                 $lastSyncUpdate = $time;
                 $timerEnd = \microtime(true);
 
+                $pools->reclaim();
+
                 Console::log("Sync tick: {$total} schedules were updated in " . ($timerEnd - $timerStart) . " seconds");
             });
 

src/Appwrite/Platform/Tasks/ScheduleExecutions.php

@@ -3,6 +3,7 @@
 namespace Appwrite\Platform\Tasks;
 
 use Appwrite\Event\Func;
+use Swoole\Coroutine as Co;
 use Utopia\Database\Database;
 use Utopia\Pools\Group;
 
@@ -28,6 +29,9 @@ class ScheduleExecutions extends ScheduleBase
 
     protected function enqueueResources(Group $pools, Database $dbForPlatform, callable $getProjectDB): void
     {
+        $queue = $pools->get('publisher')->pop();
+        $connection = $queue->getResource();
+        $queueForFunctions = new Func($connection);
         $intervalEnd = (new \DateTime())->modify('+' . self::ENQUEUE_TIMER . ' seconds');
 
         foreach ($this->schedules as $schedule) {
@@ -55,10 +59,8 @@ class ScheduleExecutions extends ScheduleBase
 
             $this->updateProjectAccess($schedule['project'], $dbForPlatform);
 
-            \go(function () use ($schedule, $delay, $data, $pools) {
-                \Co::sleep($delay);
-
-                $queueForFunctions = new Func($this->publisher);
+            \go(function () use ($queueForFunctions, $schedule, $delay, $data) {
+                Co::sleep($delay);
 
                 $queueForFunctions->setType('schedule')
                     // Set functionId instead of function as we don't have $dbForProject
@@ -81,5 +83,7 @@ class ScheduleExecutions extends ScheduleBase
 
             unset($this->schedules[$schedule['$internalId']]);
         }
+
+        $queue->reclaim();
     }
 }

src/Appwrite/Platform/Tasks/ScheduleFunctions.php

@@ -71,7 +71,10 @@ class ScheduleFunctions extends ScheduleBase
 
         foreach ($delayedExecutions as $delay => $scheduleKeys) {
             \go(function () use ($delay, $scheduleKeys, $pools, $dbForPlatform) {
-                \Co::sleep($delay); // in seconds
+                \sleep($delay); // in seconds
+
+                $queue = $pools->get('publisher')->pop();
+                $connection = $queue->getResource();
 
                 foreach ($scheduleKeys as $scheduleKey) {
                     // Ensure schedule was not deleted
@@ -83,7 +86,7 @@ class ScheduleFunctions extends ScheduleBase
 
                     $this->updateProjectAccess($schedule['project'], $dbForPlatform);
 
-                    $queueForFunctions = new Func($this->publisher);
+                    $queueForFunctions = new Func($connection);
 
                     $queueForFunctions
                         ->setType('schedule')
@@ -93,6 +96,8 @@ class ScheduleFunctions extends ScheduleBase
                         ->setProject($schedule['project'])
                         ->trigger();
                 }
+
+                $queue->reclaim();
             });
         }
 

src/Appwrite/Platform/Tasks/ScheduleMessages.php

@@ -41,7 +41,9 @@ class ScheduleMessages extends ScheduleBase
             }
 
             \go(function () use ($schedule, $pools, $dbForPlatform) {
-                $queueForMessaging = new Messaging($this->publisher);
+                $queue = $pools->get('publisher')->pop();
+                $connection = $queue->getResource();
+                $queueForMessaging = new Messaging($connection);
 
                 $this->updateProjectAccess($schedule['project'], $dbForPlatform);
 
@@ -56,6 +58,8 @@ class ScheduleMessages extends ScheduleBase
                     $schedule['$id'],
                 );
 
+                $queue->reclaim();
+
                 unset($this->schedules[$schedule['$internalId']]);
             });
         }

src/Appwrite/Platform/Workers/Deletes.php

@@ -61,7 +61,10 @@ class Deletes extends Action
             ->inject('executionRetention')
             ->inject('auditRetention')
             ->inject('log')
-            ->callback([$this, 'action']);
+            ->callback(
+                fn ($message, Document $project, Database $dbForPlatform, callable $getProjectDB, callable $getLogsDB, Device $deviceForFiles, Device $deviceForFunctions, Device $deviceForBuilds, Device $deviceForCache, CertificatesAdapter $certificates, Executor $executor, string $executionRetention, string $auditRetention, Log $log) =>
+                    $this->action($message, $project, $dbForPlatform, $getProjectDB, $getLogsDB, $deviceForFiles, $deviceForFunctions, $deviceForBuilds, $deviceForCache, $certificates, $executor, $executionRetention, $auditRetention, $log)
+            );
     }
 
     /**

src/Appwrite/Platform/Workers/StatsUsage.php

@@ -325,7 +325,7 @@ class StatsUsage extends Action
                     break;
             }
         } catch (Throwable $e) {
-            Console::error("[reducer] " . " {DateTime::now()} " . " {$project->getInternalId()} " . " {$e->getMessage()}");
+            console::error("[reducer] " . " {DateTime::now()} " . " {$project->getInternalId()} " . " {$e->getMessage()}");
         }
     }
 
@@ -344,7 +344,7 @@ class StatsUsage extends Action
                 continue;
             }
 
-            Console::log('['.DateTime::now().'] Id: '.$project->getId(). ' InternalId: '.$project->getInternalId(). ' Db: '.$project->getAttribute('database').' ReceivedAt: '.$receivedAt. ' Keys: '.$numberOfKeys);
+            console::log('['.DateTime::now().'] Id: '.$project->getId(). ' InternalId: '.$project->getInternalId(). ' Db: '.$project->getAttribute('database').' ReceivedAt: '.$receivedAt. ' Keys: '.$numberOfKeys);
 
             try {
                 foreach ($stats['keys'] ?? [] as $key => $value) {
@@ -381,7 +381,7 @@ class StatsUsage extends Action
                     }
                 }
             } catch (Exception $e) {
-                Console::error('[' . DateTime::now() . '] project [' . $project->getInternalId() . '] database [' . $project['database'] . '] ' . ' ' . $e->getMessage());
+                console::error('[' . DateTime::now() . '] project [' . $project->getInternalId() . '] database [' . $project['database'] . '] ' . ' ' . $e->getMessage());
             }
         }
 
@@ -405,7 +405,7 @@ class StatsUsage extends Action
 
     }
 
-    protected function prepareForLogsDB(Document $project, Document $stat): void
+    protected function prepareForLogsDB(Document $project, Document $stat)
     {
         if (System::getEnv('_APP_STATS_USAGE_DUAL_WRITING', 'disabled') === 'disabled') {
             return;
@@ -430,7 +430,8 @@ class StatsUsage extends Action
             return;
         }
 
-        $dbForLogs = ($this->getLogsDB)()
+        $dbForLogs = call_user_func($this->getLogsDB);
+        $dbForLogs
             ->setTenant(null)
             ->setTenantPerDocument(true);
 
@@ -445,5 +446,6 @@ class StatsUsage extends Action
         } catch (Throwable $th) {
             Console::error($th->getMessage());
         }
+        $this->register->get('pools')->get('logs')->reclaim();
     }
 }

src/Appwrite/Platform/Workers/StatsUsageDump.php

@@ -70,9 +70,9 @@ class StatsUsageDump extends Action
     ];
 
     /**
-     * @var callable(Document): Database
+     * @var callable
      */
-    protected $getLogsDB;
+    protected mixed $getLogsDB;
 
     protected array $periods = [
         '1h' => 'Y-m-d H:00',
@@ -126,10 +126,10 @@ class StatsUsageDump extends Action
                 continue;
             }
 
-            Console::log('['.DateTime::now().'] Id: '.$project->getId(). ' InternalId: '.$project->getInternalId(). ' Db: '.$project->getAttribute('database').' ReceivedAt: '.$receivedAt. ' Keys: '.$numberOfKeys);
+            console::log('['.DateTime::now().'] Id: '.$project->getId(). ' InternalId: '.$project->getInternalId(). ' Db: '.$project->getAttribute('database').' ReceivedAt: '.$receivedAt. ' Keys: '.$numberOfKeys);
 
             try {
-                /** @var Database $dbForProject */
+                /** @var \Utopia\Database\Database $dbForProject */
                 $dbForProject = $getProjectDB($project);
                 foreach ($stats['keys'] ?? [] as $key => $value) {
                     if ($value == 0) {
@@ -169,7 +169,7 @@ class StatsUsageDump extends Action
                     }
                 }
             } catch (\Exception $e) {
-                Console::error('[' . DateTime::now() . '] project [' . $project->getInternalId() . '] database [' . $project['database'] . '] ' . ' ' . $e->getMessage());
+                console::error('[' . DateTime::now() . '] project [' . $project->getInternalId() . '] database [' . $project['database'] . '] ' . ' ' . $e->getMessage());
             }
         }
     }
@@ -190,7 +190,8 @@ class StatsUsageDump extends Action
             }
         }
 
-        $dbForLogs = ($this->getLogsDB)($project);
+        /** @var \Utopia\Database\Database $dbForLogs*/
+        $dbForLogs = call_user_func($this->getLogsDB, $project);
 
         try {
             $dbForLogs->createOrUpdateDocumentsWithIncrease(
@@ -202,5 +203,7 @@ class StatsUsageDump extends Action
         } catch (\Throwable $th) {
             Console::error($th->getMessage());
         }
+
+        $this->register->get('pools')->get('logs')->reclaim();
     }
 }

src/Appwrite/PubSub/Adapter/Pool.php

@@ -1,46 +0,0 @@
-<?php
-
-namespace Appwrite\PubSub\Adapter;
-
-use Appwrite\PubSub\Adapter;
-use Utopia\Database\Exception as DatabaseException;
-use Utopia\Pools\Pool as UtopiaPool;
-
-class Pool implements Adapter
-{
-    public function __construct(private UtopiaPool $pool)
-    {
-    }
-
-    public function ping($message = null): bool
-    {
-        return $this->delegate(__FUNCTION__, \func_get_args());
-    }
-
-    public function subscribe($channels, $callback): void
-    {
-        $this->delegate(__FUNCTION__, \func_get_args());
-    }
-
-    public function publish($channel, $message): void
-    {
-        $this->delegate(__FUNCTION__, \func_get_args());
-    }
-
-    /**
-     * Forward method calls to the internal adapter instance via the pool.
-     *
-     * Required because __call() can't be used to implement abstract methods.
-     *
-     * @param string $method
-     * @param array<mixed> $args
-     * @return mixed
-     * @throws DatabaseException
-     */
-    public function delegate(string $method, array $args): mixed
-    {
-        return $this->pool->use(function (Adapter $adapter) use ($method, $args) {
-            return $adapter->{$method}(...$args);
-        });
-    }
-}

tests/resources/docker/docker-compose.yml

@@ -35,7 +35,7 @@ services:
         - VERSION=dev
     restart: unless-stopped
     ports:
-      - "9501:80"
+      - 9501:80
     networks:
       - appwrite
     labels:
@@ -52,12 +52,15 @@ services:
       - ./phpunit.xml:/usr/src/code/phpunit.xml
       - ./tests:/usr/src/code/tests
       - ./app:/usr/src/code/app
+      # - ./vendor:/usr/src/code/vendor
       - ./docs:/usr/src/code/docs
       - ./public:/usr/src/code/public
       - ./src:/usr/src/code/src
+      - ./debug:/tmp
     depends_on:
       - mariadb
       - redis
+      # - clamav
     environment:
       - _APP_COMPRESSION_MIN_SIZE_BYTES
       - _APP_ENV
@@ -352,6 +355,33 @@ services:
     volumes:
       - appwrite-redis:/data:rw
 
+  # clamav:
+  #   image: appwrite/clamav:1.2.0
+  #   container_name: appwrite-clamav
+  #   restart: unless-stopped
+  #   networks:
+  #     - appwrite
+  #   volumes:
+  #     - appwrite-uploads:/storage/uploads
+
+
+  # redis-commander:
+  #   image: rediscommander/redis-commander:latest
+  #   restart: unless-stopped
+  #   networks:
+  #   - appwrite
+  #   environment:
+  #   - REDIS_HOSTS=redis
+  #   ports:
+  #   - "8081:8081"
+
+  # webgrind:
+  #   image: 'jokkedk/webgrind:latest'
+  #   volumes:
+  #     - './debug:/tmp'
+  #   ports:
+  #     - '3001:80'
+
 networks:
   gateway:
   appwrite: