CLI scope pool use

app/cli.php

@@ -14,6 +14,7 @@ use Utopia\Cache\Cache;
 use Utopia\CLI\CLI;
 use Utopia\CLI\Console;
 use Utopia\Config\Config;
+use Utopia\Database\Adapter\Pool as PoolAdapter;
 use Utopia\Database\Database;
 use Utopia\Database\Document;
 use Utopia\Database\Validator\Authorization;
@@ -63,12 +64,8 @@ CLI::setResource('dbForPlatform', function ($pools, $cache) {
         $attempts++;
         try {
             // Prepare database connection
-            $dbAdapter = $pools
-                ->get('console')
-                ->pop()
-                ->getResource();
-
-            $dbForPlatform = new Database($dbAdapter, $cache);
+            $adapter = new PoolAdapter($pools->get('console'));
+            $dbForPlatform = new Database($adapter, $cache);
 
             $dbForPlatform
                 ->setNamespace('_console')
@@ -86,7 +83,6 @@ 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);
@@ -136,12 +132,8 @@ CLI::setResource('getProjectDB', function (Group $pools, Database $dbForPlatform
             return $database;
         }
 
-        $dbAdapter = $pools
-            ->get($dsn->getHost())
-            ->pop()
-            ->getResource();
-
-        $database = new Database($dbAdapter, $cache);
+        $adapter = new PoolAdapter($pools->get($dsn->getHost()));
+        $database = new Database($adapter, $cache);
         $databases[$dsn->getHost()] = $database;
         $sharedTables = \explode(',', System::getEnv('_APP_DATABASE_SHARED_TABLES', ''));
 
@@ -173,15 +165,8 @@ CLI::setResource('getLogsDB', function (Group $pools, Cache $cache) {
             return $database;
         }
 
-        $dbAdapter = $pools
-            ->get('logs')
-            ->pop()
-            ->getResource();
-
-        $database = new Database(
-            $dbAdapter,
-            $cache
-        );
+        $adapter = new PoolAdapter($pools->get('logs'));
+        $database = new Database($adapter, $cache);
 
         $database
             ->setSharedTables(true)

src/Appwrite/Platform/Tasks/ScheduleBase.php

@@ -132,8 +132,6 @@ 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());
@@ -198,8 +196,6 @@ 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

@@ -6,6 +6,7 @@ use Appwrite\Event\Func;
 use Swoole\Coroutine as Co;
 use Utopia\Database\Database;
 use Utopia\Pools\Group;
+use Utopia\Queue\Publisher;
 
 class ScheduleExecutions extends ScheduleBase
 {
@@ -29,9 +30,6 @@ 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) {
@@ -59,21 +57,25 @@ class ScheduleExecutions extends ScheduleBase
 
             $this->updateProjectAccess($schedule['project'], $dbForPlatform);
 
-            \go(function () use ($queueForFunctions, $schedule, $delay, $data) {
+            \go(function () use ($schedule, $delay, $data, $pools) {
                 Co::sleep($delay);
 
-                $queueForFunctions->setType('schedule')
-                    // Set functionId instead of function as we don't have $dbForProject
-                    // TODO: Refactor to use function instead of functionId
-                    ->setFunctionId($schedule['resource']['functionId'])
-                    ->setExecution($schedule['resource'])
-                    ->setMethod($data['method'] ?? 'POST')
-                    ->setPath($data['path'] ?? '/')
-                    ->setHeaders($data['headers'] ?? [])
-                    ->setBody($data['body'] ?? '')
-                    ->setProject($schedule['project'])
-                    ->setUserId($data['userId'] ?? '')
-                    ->trigger();
+                $pools->get('publisher')->use(function(Publisher $publisher) use ($schedule, $data) {
+                    $queueForFunctions = new Func($publisher);
+
+                    $queueForFunctions->setType('schedule')
+                        // Set functionId instead of function as we don't have $dbForProject
+                        // TODO: Refactor to use function instead of functionId
+                        ->setFunctionId($schedule['resource']['functionId'])
+                        ->setExecution($schedule['resource'])
+                        ->setMethod($data['method'] ?? 'POST')
+                        ->setPath($data['path'] ?? '/')
+                        ->setHeaders($data['headers'] ?? [])
+                        ->setBody($data['body'] ?? '')
+                        ->setProject($schedule['project'])
+                        ->setUserId($data['userId'] ?? '')
+                        ->trigger();
+                });
             });
 
             $dbForPlatform->deleteDocument(
@@ -82,8 +84,6 @@ class ScheduleExecutions extends ScheduleBase
             );
 
             unset($this->schedules[$schedule['$internalId']]);
-        }
-
-        $queue->reclaim();
+}
     }
 }

src/Appwrite/Platform/Tasks/ScheduleFunctions.php

@@ -8,6 +8,7 @@ use Utopia\CLI\Console;
 use Utopia\Database\Database;
 use Utopia\Database\DateTime;
 use Utopia\Pools\Group;
+use Utopia\Queue\Publisher;
 
 class ScheduleFunctions extends ScheduleBase
 {
@@ -73,31 +74,29 @@ class ScheduleFunctions extends ScheduleBase
             \go(function () use ($delay, $scheduleKeys, $pools, $dbForPlatform) {
                 \sleep($delay); // in seconds
 
-                $queue = $pools->get('publisher')->pop();
-                $connection = $queue->getResource();
 
-                foreach ($scheduleKeys as $scheduleKey) {
-                    // Ensure schedule was not deleted
-                    if (!\array_key_exists($scheduleKey, $this->schedules)) {
-                        return;
+                    foreach ($scheduleKeys as $scheduleKey) {
+                        // Ensure schedule was not deleted
+                        if (!\array_key_exists($scheduleKey, $this->schedules)) {
+                            return;
+                        }
+
+                        $schedule = $this->schedules[$scheduleKey];
+
+                        $this->updateProjectAccess($schedule['project'], $dbForPlatform);
+
+                        $pools->get('publisher')->use(function(Publisher $publisher) use ($schedule) {
+                            $queueForFunctions = new Func($publisher);
+    
+                            $queueForFunctions
+                                ->setType('schedule')
+                                ->setFunction($schedule['resource'])
+                                ->setMethod('POST')
+                                ->setPath('/')
+                                ->setProject($schedule['project'])
+                                ->trigger();
+                        });
                     }
-
-                    $schedule = $this->schedules[$scheduleKey];
-
-                    $this->updateProjectAccess($schedule['project'], $dbForPlatform);
-
-                    $queueForFunctions = new Func($connection);
-
-                    $queueForFunctions
-                        ->setType('schedule')
-                        ->setFunction($schedule['resource'])
-                        ->setMethod('POST')
-                        ->setPath('/')
-                        ->setProject($schedule['project'])
-                        ->trigger();
-                }
-
-                $queue->reclaim();
             });
         }
 

src/Appwrite/Platform/Tasks/ScheduleMessages.php

@@ -5,6 +5,7 @@ namespace Appwrite\Platform\Tasks;
 use Appwrite\Event\Messaging;
 use Utopia\Database\Database;
 use Utopia\Pools\Group;
+use Utopia\Queue\Publisher;
 
 class ScheduleMessages extends ScheduleBase
 {
@@ -41,25 +42,23 @@ class ScheduleMessages extends ScheduleBase
             }
 
             \go(function () use ($schedule, $pools, $dbForPlatform) {
-                $queue = $pools->get('publisher')->pop();
-                $connection = $queue->getResource();
-                $queueForMessaging = new Messaging($connection);
+                $pools->get('publisher')->use(function(Publisher $publisher) use ($schedule, $dbForPlatform) {
+                    $queueForMessaging = new Messaging($publisher);
 
-                $this->updateProjectAccess($schedule['project'], $dbForPlatform);
+                    $this->updateProjectAccess($schedule['project'], $dbForPlatform);
 
-                $queueForMessaging
-                    ->setType(MESSAGE_SEND_TYPE_EXTERNAL)
-                    ->setMessageId($schedule['resourceId'])
-                    ->setProject($schedule['project'])
-                    ->trigger();
+                    $queueForMessaging
+                        ->setType(MESSAGE_SEND_TYPE_EXTERNAL)
+                        ->setMessageId($schedule['resourceId'])
+                        ->setProject($schedule['project'])
+                        ->trigger();
+                });
 
                 $dbForPlatform->deleteDocument(
                     'schedules',
                     $schedule['$id'],
                 );
 
-                $queue->reclaim();
-
                 unset($this->schedules[$schedule['$internalId']]);
             });
         }