Port Deletes and Builds worker

app/cli.php

@@ -8,6 +8,10 @@ use Appwrite\Event\Certificate;
 use Appwrite\Event\Database as EventDatabase;
 use Appwrite\Event\Func;
 use Appwrite\Event\Audit;
+use Appwrite\Event\Build;
+use Appwrite\Event\Event;
+use Appwrite\Event\Mail;
+use Appwrite\Event\Phone;
 use Appwrite\Platform\Appwrite;
 use Utopia\CLI\CLI;
 use Utopia\Database\Validator\Authorization;
@@ -149,27 +153,36 @@ CLI::setResource('influxdb', function (Registry $register) {
 CLI::setResource('queue', function (Group $pools) {
     return $pools->get('queue')->pop()->getResource();
 }, ['pools']);
-
-CLI::setResource('deletes', function (Connection $queue) {
-    return new Delete($queue);
+CLI::setResource('messaging', function (Connection $queue) {
+    return new Phone($queue);
 }, ['queue']);
-
-CLI::setResource('queueForFunctions', function (Connection $queue) {
-    return new Func($queue);
+CLI::setResource('mails', function (Connection $queue) {
+    return new Mail($queue);
 }, ['queue']);
-
-CLI::setResource('audits', function (Connection $queue) {
-    return new Audit($queue);
+CLI::setResource('builds', function (Connection $queue) {
+    return new Build($queue);
 }, ['queue']);
-
-CLI::setResource('certificates', function (Connection $queue) {
-    return new Certificate($queue);
-}, ['queue']);
-
 CLI::setResource('database', function (Connection $queue) {
     return new EventDatabase($queue);
 }, ['queue']);
-
+CLI::setResource('deletes', function (Connection $queue) {
+    return new Delete($queue);
+}, ['queue']);
+CLI::setResource('events', function (Connection $queue) {
+    return new Event('', '', $queue);
+}, ['queue']);
+CLI::setResource('audits', function (Connection $queue) {
+    return new Audit($queue);
+}, ['queue']);
+CLI::setResource('events', function (Connection $queue) {
+    return new Event('', '', $queue);
+}, ['queue']);
+CLI::setResource('queueForFunctions', function (Connection $queue) {
+    return new Func($queue);
+}, ['queue']);
+CLI::setResource('certificates', function (Connection $queue) {
+    return new Certificate($queue);
+}, ['queue']);
 CLI::setResource('logError', function (Registry $register) {
     return function (Throwable $error, string $namespace, string $action) use ($register) {
         $logger = $register->get('logger');

app/controllers/api/account.php

@@ -1291,7 +1291,6 @@ App::get('/v1/account')
     ->inject('response')
     ->inject('user')
     ->action(function (Response $response, Document $user) {
-
         $response->dynamic($user, Response::MODEL_ACCOUNT);
     });
 

app/controllers/api/functions.php

@@ -634,7 +634,8 @@ App::post('/v1/functions/:functionId/deployments')
     ->inject('deviceFunctions')
     ->inject('deviceLocal')
     ->inject('dbForConsole')
-    ->action(function (string $functionId, string $entrypoint, mixed $code, bool $activate, Request $request, Response $response, Database $dbForProject, Event $events, Document $project, Device $deviceFunctions, Device $deviceLocal, Database $dbForConsole) {
+    ->inject('builds')
+    ->action(function (string $functionId, string $entrypoint, mixed $code, bool $activate, Request $request, Response $response, Database $dbForProject, Event $events, Document $project, Device $deviceFunctions, Device $deviceLocal, Database $dbForConsole, Build $builds) {
 
         $function = $dbForProject->getDocument('functions', $functionId);
 
@@ -754,8 +755,7 @@ App::post('/v1/functions/:functionId/deployments')
             }
 
             // Start the build
-            $buildEvent = new Build();
-            $buildEvent
+            $builds
                 ->setType(BUILD_TYPE_DEPLOYMENT)
                 ->setResource($function)
                 ->setDeployment($deployment)
@@ -1001,7 +1001,8 @@ App::post('/v1/functions/:functionId/deployments/:deploymentId/builds/:buildId')
     ->inject('dbForProject')
     ->inject('project')
     ->inject('events')
-    ->action(function (string $functionId, string $deploymentId, string $buildId, Response $response, Database $dbForProject, Document $project, Event $events) {
+    ->inject('builds')
+    ->action(function (string $functionId, string $deploymentId, string $buildId, Response $response, Database $dbForProject, Document $project, Event $events, Build $builds) {
 
         $function = $dbForProject->getDocument('functions', $functionId);
         $deployment = $dbForProject->getDocument('deployments', $deploymentId);
@@ -1029,8 +1030,7 @@ App::post('/v1/functions/:functionId/deployments/:deploymentId/builds/:buildId')
             ->setParam('deploymentId', $deployment->getId());
 
         // Retry the build
-        $buildEvent = new Build();
-        $buildEvent
+        $builds
             ->setType(BUILD_TYPE_RETRY)
             ->setResource($function)
             ->setDeployment($deployment)

app/init.php

@@ -69,6 +69,7 @@ use Utopia\Pools\Group;
 use Utopia\Pools\Pool;
 use Ahc\Jwt\JWT;
 use Ahc\Jwt\JWTException;
+use Appwrite\Event\Build;
 use Appwrite\Event\Certificate;
 use Appwrite\Event\Func;
 use MaxMind\Db\Reader;
@@ -858,18 +859,21 @@ App::setResource('messaging', function (Connection $queue) {
 App::setResource('mails', function (Connection $queue) {
     return new Mail($queue);
 }, ['queue']);
+App::setResource('builds', function (Connection $queue) {
+    return new Build($queue);
+}, ['queue']);
 App::setResource('database', function (Connection $queue) {
     return new EventDatabase($queue);
-}, ['pools']);
+}, ['queue']);
 App::setResource('deletes', function (Connection $queue) {
     return new Delete($queue);
 }, ['queue']);
 App::setResource('events', function (Connection $queue) {
     return new Event('', '', $queue);
-}, ['pools']);
+}, ['queue']);
 App::setResource('audits', function (Connection $queue) {
     return new Audit($queue);
-}, ['pools']);
+}, ['queue']);
 App::setResource('events', function (Connection $queue) {
     return new Event('', '', $queue);
 }, ['queue']);

app/worker.php

@@ -4,9 +4,13 @@ require_once __DIR__ . '/init.php';
 
 use Appwrite\Event\Event;
 use Appwrite\Event\Audit;
+use Appwrite\Event\Build;
 use Appwrite\Event\Certificate;
 use Appwrite\Event\Database as EventDatabase;
+use Appwrite\Event\Delete;
 use Appwrite\Event\Func;
+use Appwrite\Event\Mail;
+use Appwrite\Event\Phone;
 use Swoole\Runtime;
 use Utopia\App;
 use Utopia\Cache\Adapter\Sharding;
@@ -15,6 +19,7 @@ use Utopia\CLI\Console;
 use Utopia\Config\Config;
 use Utopia\Database\Database;
 use Utopia\Database\Document;
+use Utopia\DSN\DSN;
 use Utopia\Queue\Adapter\Swoole;
 use Utopia\Queue\Message;
 use Utopia\Queue\Server;
@@ -22,6 +27,15 @@ use Utopia\Registry\Registry;
 use Utopia\Logger\Log;
 use Utopia\Logger\Logger;
 use Utopia\Pools\Group;
+use Utopia\Queue\Connection;
+use Utopia\Storage\Device;
+use Utopia\Storage\Device\Backblaze;
+use Utopia\Storage\Device\DOSpaces;
+use Utopia\Storage\Device\Linode;
+use Utopia\Storage\Device\Local;
+use Utopia\Storage\Device\S3;
+use Utopia\Storage\Device\Wasabi;
+use Utopia\Storage\Storage;
 
 Runtime::enableCoroutine(SWOOLE_HOOK_ALL);
 
@@ -86,16 +100,39 @@ Server::setResource('database', function (Registry $register) {
     );
 }, ['register']);
 
-Server::setResource('queueForFunctions', function (Registry $register) {
-    $pools = $register->get('pools');
-    return new Func(
-        $pools
-            ->get('queue')
-            ->pop()
-            ->getResource()
-    );
-}, ['register']);
-
+Server::setResource('queue', function (Group $pools) {
+    return $pools->get('queue')->pop()->getResource();
+}, ['pools']);
+Server::setResource('messaging', function (Connection $queue) {
+    return new Phone($queue);
+}, ['queue']);
+Server::setResource('mails', function (Connection $queue) {
+    return new Mail($queue);
+}, ['queue']);
+Server::setResource('builds', function (Connection $queue) {
+    return new Build($queue);
+}, ['queue']);
+Server::setResource('database', function (Connection $queue) {
+    return new EventDatabase($queue);
+}, ['queue']);
+Server::setResource('deletes', function (Connection $queue) {
+    return new Delete($queue);
+}, ['queue']);
+Server::setResource('events', function (Connection $queue) {
+    return new Event('', '', $queue);
+}, ['queue']);
+Server::setResource('audits', function (Connection $queue) {
+    return new Audit($queue);
+}, ['queue']);
+Server::setResource('events', function (Connection $queue) {
+    return new Event('', '', $queue);
+}, ['queue']);
+Server::setResource('queueForFunctions', function (Connection $queue) {
+    return new Func($queue);
+}, ['queue']);
+Server::setResource('certificates', function (Connection $queue) {
+    return new Certificate($queue);
+}, ['queue']);
 Server::setResource('events', function (Registry $register) {
     $pools = $register->get('pools');
     return new Event(
@@ -197,3 +234,111 @@ $server
         Console::error('[Error] File: ' . $error->getFile());
         Console::error('[Error] Line: ' . $error->getLine());
     });
+
+/**
+ * Get Console DB
+ *
+ * @returns Cache
+ */
+function getCache(): Cache
+{
+    global $register;
+
+    $pools = $register->get('pools');
+    /** @var \Utopia\Pools\Group $pools */
+
+    $list = Config::getParam('pools-cache', []);
+    $adapters = [];
+
+    foreach ($list as $value) {
+        $adapters[] = $pools
+            ->get($value)
+            ->pop()
+            ->getResource();
+    }
+
+    return new Cache(new Sharding($adapters));
+}
+
+/**
+ * Get Console DB
+ *
+ * @returns Database
+ */
+function getConsoleDB(): Database
+{
+    global $register;
+
+    /** @var \Utopia\Pools\Group $pools */
+    $pools = $register->get('pools');
+
+    $dbAdapter = $pools
+        ->get('console')
+        ->pop()
+        ->getResource();
+
+    $database = new Database($dbAdapter, getCache());
+
+    $database->setNamespace('console');
+
+    return $database;
+}
+
+/**
+ * Get Project DB
+ *
+ * @param Document $project
+ * @returns Database
+ */
+function getProjectDB(Document $project): Database
+{
+    global $register;
+
+    /** @var \Utopia\Pools\Group $pools */
+    $pools = $register->get('pools');
+
+    if ($project->isEmpty() || $project->getId() === 'console') {
+        return getConsoleDB();
+    }
+
+    $dbAdapter = $pools
+        ->get($project->getAttribute('database'))
+        ->pop()
+        ->getResource();
+
+    $database = new Database($dbAdapter, getCache());
+    $database->setNamespace('_' . $project->getInternalId());
+
+    return $database;
+}
+
+
+/**
+ * Get Functions Storage Device
+ * @param string $projectId of the project
+ * @return Device
+ */
+function getFunctionsDevice($projectId): Device
+{
+    return getDevice(APP_STORAGE_FUNCTIONS . '/app-' . $projectId);
+}
+
+/**
+ * Get Files Storage Device
+ * @param string $projectId of the project
+ * @return Device
+ */
+function getFilesDevice($projectId): Device
+{
+    return getDevice(APP_STORAGE_UPLOADS . '/app-' . $projectId);
+}
+
+/**
+ * Get Builds Storage Device
+ * @param string $projectId of the project
+ * @return Device
+ */
+function getBuildsDevice($projectId): Device
+{
+    return getDevice(APP_STORAGE_BUILDS . '/app-' . $projectId);
+}

app/workers/builds.php

@@ -1,5 +1,7 @@
 <?php
 
+require_once __DIR__ . '/../worker.php';
+
 use Appwrite\Event\Event;
 use Appwrite\Event\Func;
 use Appwrite\Messaging\Adapter\Realtime;
@@ -15,147 +17,200 @@ use Utopia\DSN\DSN;
 use Utopia\Database\Document;
 use Utopia\Config\Config;
 use Utopia\Database\Validator\Authorization;
+use Utopia\Queue\Message;
 use Utopia\Storage\Storage;
 
-require_once __DIR__ . '/../init.php';
+Authorization::disable();
+Authorization::setDefaultStatus(false);
 
-Console::title('Builds V1 Worker');
-Console::success(APP_NAME . ' build worker v1 has started');
-
-// TODO: Executor should return appropriate response codes.
-class BuildsV1 extends Worker
+function buildDeployment(Document $project, Document $function, Document $deployment)
 {
-    private ?Executor $executor = null;
+    global $register;
+    $executor = new Executor(App::getEnv('_APP_EXECUTOR_HOST'));
 
-    public function getName(): string
-    {
-        return "builds";
+    $dbForProject = getProjectDB($project);
+
+    $function = $dbForProject->getDocument('functions', $function->getId());
+    if ($function->isEmpty()) {
+        throw new Exception('Function not found', 404);
     }
 
-    public function init(): void
-    {
-        $this->executor = new Executor(App::getEnv('_APP_EXECUTOR_HOST'));
+    $deployment = $dbForProject->getDocument('deployments', $deployment->getId());
+    if ($deployment->isEmpty()) {
+        throw new Exception('Deployment not found', 404);
     }
 
-    public function run(): void
-    {
-        $type = $this->args['type'] ?? '';
-        $project = new Document($this->args['project'] ?? []);
-        $resource = new Document($this->args['resource'] ?? []);
-        $deployment = new Document($this->args['deployment'] ?? []);
-
-        switch ($type) {
-            case BUILD_TYPE_DEPLOYMENT:
-            case BUILD_TYPE_RETRY:
-                Console::info('Creating build for deployment: ' . $deployment->getId());
-                $this->buildDeployment($project, $resource, $deployment);
-                break;
-
-            default:
-                throw new \Exception('Invalid build type');
-                break;
-        }
+    $runtimes = Config::getParam('runtimes', []);
+    $key = $function->getAttribute('runtime');
+    $runtime = isset($runtimes[$key]) ? $runtimes[$key] : null;
+    if (\is_null($runtime)) {
+        throw new Exception('Runtime "' . $function->getAttribute('runtime', '') . '" is not supported');
     }
 
-    protected function buildDeployment(Document $project, Document $function, Document $deployment)
-    {
-        global $register;
+    $connection = App::getEnv('_APP_CONNECTIONS_STORAGE', ''); /** @TODO : move this to the registry or someplace else */
+    $device = Storage::DEVICE_LOCAL;
+    try {
+        $dsn = new DSN($connection);
+        $device = $dsn->getScheme();
+    } catch (\Exception $e) {
+        Console::error($e->getMessage() . 'Invalid DSN. Defaulting to Local device.');
+    }
 
-        $dbForProject = $this->getProjectDB($project);
+    $buildId = $deployment->getAttribute('buildId', '');
+    $startTime = DateTime::now();
+    if (empty($buildId)) {
+        $buildId = ID::unique();
+        $build = $dbForProject->createDocument('builds', new Document([
+            '$id' => $buildId,
+            '$permissions' => [],
+            'startTime' => $startTime,
+            'deploymentId' => $deployment->getId(),
+            'status' => 'processing',
+            'outputPath' => '',
+            'runtime' => $function->getAttribute('runtime'),
+            'source' => $deployment->getAttribute('path'),
+            'sourceType' => $device,
+            'stdout' => '',
+            'stderr' => '',
+            'endTime' => null,
+            'duration' => 0
+        ]));
+        $deployment->setAttribute('buildId', $buildId);
+        $deployment = $dbForProject->updateDocument('deployments', $deployment->getId(), $deployment);
+    } else {
+        $build = $dbForProject->getDocument('builds', $buildId);
+    }
 
-        $function = $dbForProject->getDocument('functions', $function->getId());
-        if ($function->isEmpty()) {
-            throw new Exception('Function not found', 404);
+    /** Request the executor to build the code... */
+    $build->setAttribute('status', 'building');
+    $build = $dbForProject->updateDocument('builds', $buildId, $build);
+
+    /** Trigger Webhook */
+    $deploymentModel = new Deployment();
+
+    $pools = $register->get('pools');
+    $connection = $pools->get('queue')->pop()->getResource();
+
+    $deploymentUpdate = new Event(Event::WEBHOOK_QUEUE_NAME, Event::WEBHOOK_CLASS_NAME, $connection);
+    $deploymentUpdate
+        ->setProject($project)
+        ->setEvent('functions.[functionId].deployments.[deploymentId].update')
+        ->setParam('functionId', $function->getId())
+        ->setParam('deploymentId', $deployment->getId())
+        ->setPayload($deployment->getArrayCopy(array_keys($deploymentModel->getRules())))
+        ->trigger();
+
+    /** Trigger Functions */
+    $pools = $register->get('pools');
+    $connection = $pools->get('queue')->pop();
+
+    $functions = new Func($connection->getResource());
+    $functions
+        ->from($deploymentUpdate)
+        ->trigger();
+
+    $connection->reclaim();
+
+    /** Trigger Realtime */
+    $allEvents = Event::generateEvents('functions.[functionId].deployments.[deploymentId].update', [
+        'functionId' => $function->getId(),
+        'deploymentId' => $deployment->getId()
+    ]);
+    $target = Realtime::fromPayload(
+        // Pass first, most verbose event pattern
+        event: $allEvents[0],
+        payload: $build,
+        project: $project
+    );
+
+    Realtime::send(
+        projectId: 'console',
+        payload: $build->getArrayCopy(),
+        events: $allEvents,
+        channels: $target['channels'],
+        roles: $target['roles']
+    );
+
+    $source = $deployment->getAttribute('path');
+
+    $vars = array_reduce($function->getAttribute('vars', []), function (array $carry, Document $var) {
+        $carry[$var->getAttribute('key')] = $var->getAttribute('value');
+        return $carry;
+    }, []);
+
+    try {
+        $response = $executor->createRuntime(
+            projectId: $project->getId(),
+            deploymentId: $deployment->getId(),
+            source: $source,
+            image: $runtime['image'],
+            remove: true,
+            entrypoint: $deployment->getAttribute('entrypoint'),
+            workdir: '/usr/code',
+            destination: APP_STORAGE_BUILDS . "/app-{$project->getId()}",
+            variables: $vars,
+            commands: [
+                'sh', '-c',
+                'tar -zxf /tmp/code.tar.gz -C /usr/code && \
+                cd /usr/local/src/ && ./build.sh'
+            ]
+        );
+
+        $endTime = new \DateTime();
+        $endTime->setTimestamp($response['endTimeUnix']);
+
+        /** Update the build document */
+        $build->setAttribute('endTime', DateTime::format($endTime));
+        $build->setAttribute('duration', \intval($response['duration']));
+        $build->setAttribute('status', $response['status']);
+        $build->setAttribute('outputPath', $response['outputPath']);
+        $build->setAttribute('stderr', $response['stderr']);
+        $build->setAttribute('stdout', $response['stdout']);
+
+        /* Also update the deployment buildTime */
+        $deployment->setAttribute('buildTime', $response['duration']);
+
+        Console::success("Build id: $buildId created");
+
+        $function->setAttribute('scheduleUpdatedAt', DateTime::now());
+
+        /** Set auto deploy */
+        if ($deployment->getAttribute('activate') === true) {
+            $function->setAttribute('deployment', $deployment->getId());
+            $function = $dbForProject->updateDocument('functions', $function->getId(), $function);
         }
 
-        $deployment = $dbForProject->getDocument('deployments', $deployment->getId());
-        if ($deployment->isEmpty()) {
-            throw new Exception('Deployment not found', 404);
-        }
+        /** Update function schedule */
+        $dbForConsole = getConsoleDB();
+        $schedule = $dbForConsole->getDocument('schedules', $function->getAttribute('scheduleId'));
+        $schedule->setAttribute('resourceUpdatedAt', $function->getAttribute('scheduleUpdatedAt'));
 
-        $runtimes = Config::getParam('runtimes', []);
-        $key = $function->getAttribute('runtime');
-        $runtime = isset($runtimes[$key]) ? $runtimes[$key] : null;
-        if (\is_null($runtime)) {
-            throw new Exception('Runtime "' . $function->getAttribute('runtime', '') . '" is not supported');
-        }
+        $schedule
+            ->setAttribute('schedule', $function->getAttribute('schedule'))
+            ->setAttribute('active', !empty($function->getAttribute('schedule')) && !empty($function->getAttribute('deployment')));
 
-        $connection = App::getEnv('_APP_CONNECTIONS_STORAGE', ''); /** @TODO : move this to the registry or someplace else */
-        $device = Storage::DEVICE_LOCAL;
-        try {
-            $dsn = new DSN($connection);
-            $device = $dsn->getScheme();
-        } catch (\Exception $e) {
-            Console::error($e->getMessage() . 'Invalid DSN. Defaulting to Local device.');
-        }
 
-        $buildId = $deployment->getAttribute('buildId', '');
-        $startTime = DateTime::now();
-        if (empty($buildId)) {
-            $buildId = ID::unique();
-            $build = $dbForProject->createDocument('builds', new Document([
-                '$id' => $buildId,
-                '$permissions' => [],
-                'startTime' => $startTime,
-                'deploymentId' => $deployment->getId(),
-                'status' => 'processing',
-                'outputPath' => '',
-                'runtime' => $function->getAttribute('runtime'),
-                'source' => $deployment->getAttribute('path'),
-                'sourceType' => $device,
-                'stdout' => '',
-                'stderr' => '',
-                'endTime' => null,
-                'duration' => 0
-            ]));
-            $deployment->setAttribute('buildId', $buildId);
-            $deployment = $dbForProject->updateDocument('deployments', $deployment->getId(), $deployment);
-        } else {
-            $build = $dbForProject->getDocument('builds', $buildId);
-        }
-
-        /** Request the executor to build the code... */
-        $build->setAttribute('status', 'building');
+        Authorization::skip(fn () => $dbForConsole->updateDocument('schedules', $schedule->getId(), $schedule));
+    } catch (\Throwable $th) {
+        $endTime = DateTime::now();
+        $interval = (new \DateTime($endTime))->diff(new \DateTime($startTime));
+        $build->setAttribute('endTime', $endTime);
+        $build->setAttribute('duration', $interval->format('%s') + 0);
+        $build->setAttribute('status', 'failed');
+        $build->setAttribute('stderr', $th->getMessage());
+        Console::error($th->getMessage());
+    } finally {
         $build = $dbForProject->updateDocument('builds', $buildId, $build);
 
-        /** Trigger Webhook */
-        $deploymentModel = new Deployment();
-
-        $pools = $register->get('pools');
-        $connection = $pools->get('queue')->pop()->getResource();
-
-        $deploymentUpdate = new Event(Event::WEBHOOK_QUEUE_NAME, Event::WEBHOOK_CLASS_NAME, $connection);
-        $deploymentUpdate
-            ->setProject($project)
-            ->setEvent('functions.[functionId].deployments.[deploymentId].update')
-            ->setParam('functionId', $function->getId())
-            ->setParam('deploymentId', $deployment->getId())
-            ->setPayload($deployment->getArrayCopy(array_keys($deploymentModel->getRules())))
-            ->trigger();
-
-        /** Trigger Functions */
-        $pools = $register->get('pools');
-        $connection = $pools->get('queue')->pop();
-
-        $functions = new Func($connection->getResource());
-        $functions
-            ->from($deploymentUpdate)
-            ->trigger();
-
-        $connection->reclaim();
-
-        /** Trigger Realtime */
-        $allEvents = Event::generateEvents('functions.[functionId].deployments.[deploymentId].update', [
-            'functionId' => $function->getId(),
-            'deploymentId' => $deployment->getId()
-        ]);
+        /**
+         * Send realtime Event
+         */
         $target = Realtime::fromPayload(
             // Pass first, most verbose event pattern
             event: $allEvents[0],
             payload: $build,
             project: $project
         );
-
         Realtime::send(
             projectId: 'console',
             payload: $build->getArrayCopy(),
@@ -164,113 +219,50 @@ class BuildsV1 extends Worker
             roles: $target['roles']
         );
 
-        $source = $deployment->getAttribute('path');
-
-        $vars = array_reduce($function->getAttribute('vars', []), function (array $carry, Document $var) {
-            $carry[$var->getAttribute('key')] = $var->getAttribute('value');
-            return $carry;
-        }, []);
-
-        try {
-            $response = $this->executor->createRuntime(
-                projectId: $project->getId(),
-                deploymentId: $deployment->getId(),
-                source: $source,
-                image: $runtime['image'],
-                remove: true,
-                entrypoint: $deployment->getAttribute('entrypoint'),
-                workdir: '/usr/code',
-                destination: APP_STORAGE_BUILDS . "/app-{$project->getId()}",
-                variables: $vars,
-                commands: [
-                    'sh', '-c',
-                    'tar -zxf /tmp/code.tar.gz -C /usr/code && \
-                    cd /usr/local/src/ && ./build.sh'
-                ]
-            );
-
-            $endTime = new \DateTime();
-            $endTime->setTimestamp($response['endTimeUnix']);
-
-            /** Update the build document */
-            $build->setAttribute('endTime', DateTime::format($endTime));
-            $build->setAttribute('duration', \intval($response['duration']));
-            $build->setAttribute('status', $response['status']);
-            $build->setAttribute('outputPath', $response['outputPath']);
-            $build->setAttribute('stderr', $response['stderr']);
-            $build->setAttribute('stdout', $response['stdout']);
-
-            /* Also update the deployment buildTime */
-            $deployment->setAttribute('buildTime', $response['duration']);
-
-            Console::success("Build id: $buildId created");
-
-            $function->setAttribute('scheduleUpdatedAt', DateTime::now());
-
-            /** Set auto deploy */
-            if ($deployment->getAttribute('activate') === true) {
-                $function->setAttribute('deployment', $deployment->getId());
-                $function = $dbForProject->updateDocument('functions', $function->getId(), $function);
-            }
-
-            /** Update function schedule */
-            $dbForConsole = $this->getConsoleDB();
-            $schedule = $dbForConsole->getDocument('schedules', $function->getAttribute('scheduleId'));
-            $schedule->setAttribute('resourceUpdatedAt', $function->getAttribute('scheduleUpdatedAt'));
-
-            $schedule
-                ->setAttribute('schedule', $function->getAttribute('schedule'))
-                ->setAttribute('active', !empty($function->getAttribute('schedule')) && !empty($function->getAttribute('deployment')));
-
-
-            Authorization::skip(fn () => $dbForConsole->updateDocument('schedules', $schedule->getId(), $schedule));
-        } catch (\Throwable $th) {
-            $endTime = DateTime::now();
-            $interval = (new \DateTime($endTime))->diff(new \DateTime($startTime));
-            $build->setAttribute('endTime', $endTime);
-            $build->setAttribute('duration', $interval->format('%s') + 0);
-            $build->setAttribute('status', 'failed');
-            $build->setAttribute('stderr', $th->getMessage());
-            Console::error($th->getMessage());
-        } finally {
-            $build = $dbForProject->updateDocument('builds', $buildId, $build);
-
-            /**
-             * Send realtime Event
-             */
-            $target = Realtime::fromPayload(
-                // Pass first, most verbose event pattern
-                event: $allEvents[0],
-                payload: $build,
-                project: $project
-            );
-            Realtime::send(
-                projectId: 'console',
-                payload: $build->getArrayCopy(),
-                events: $allEvents,
-                channels: $target['channels'],
-                roles: $target['roles']
-            );
-
-            /** Update usage stats */
-            if (App::getEnv('_APP_USAGE_STATS', 'enabled') === 'enabled') {
-                $statsd = $register->get('statsd');
-                $usage = new Stats($statsd);
-                $usage
-                    ->setParam('projectInternalId', $project->getInternalId())
-                    ->setParam('projectId', $project->getId())
-                    ->setParam('functionId', $function->getId())
-                    ->setParam('builds.{scope}.compute', 1)
-                    ->setParam('buildStatus', $build->getAttribute('status', ''))
-                    ->setParam('buildTime', $build->getAttribute('duration'))
-                    ->setParam('networkRequestSize', 0)
-                    ->setParam('networkResponseSize', 0)
-                    ->submit();
-            }
+        /** Update usage stats */
+        if (App::getEnv('_APP_USAGE_STATS', 'enabled') === 'enabled') {
+            $statsd = $register->get('statsd');
+            $usage = new Stats($statsd);
+            $usage
+                ->setParam('projectInternalId', $project->getInternalId())
+                ->setParam('projectId', $project->getId())
+                ->setParam('functionId', $function->getId())
+                ->setParam('builds.{scope}.compute', 1)
+                ->setParam('buildStatus', $build->getAttribute('status', ''))
+                ->setParam('buildTime', $build->getAttribute('duration'))
+                ->setParam('networkRequestSize', 0)
+                ->setParam('networkResponseSize', 0)
+                ->submit();
         }
     }
-
-    public function shutdown(): void
-    {
-    }
 }
+
+$server->job()
+    ->inject('message')
+    ->action(function (Message $message) {
+        $payload = $message->getPayload() ?? [];
+
+        if (empty($payload)) {
+            throw new Exception('Missing payload');
+        }
+
+        $type = $payload['type'] ?? '';
+        $project = new Document($payload['project'] ?? []);
+        $resource = new Document($payload['resource'] ?? []);
+        $deployment = new Document($payload['deployment'] ?? []);
+
+        switch ($type) {
+            case BUILD_TYPE_DEPLOYMENT:
+            case BUILD_TYPE_RETRY:
+                Console::info('Creating build for deployment: ' . $deployment->getId());
+                buildDeployment($project, $resource, $deployment);
+                break;
+
+            default:
+                throw new \Exception('Invalid build type');
+                break;
+        }
+    });
+
+$server->workerStart();
+$server->start();

app/workers/databases.php

@@ -25,56 +25,6 @@ Authorization::setDefaultStatus(false);
 const DATABASE_PROJECT = 'project';
 const DATABASE_CONSOLE = 'console';
 
-function getCache(): Cache
-{
-    global $register;
-
-    $pools = $register->get('pools');
-    /** @var \Utopia\Pools\Group $pools */
-
-    $list = Config::getParam('pools-cache', []);
-    $adapters = [];
-
-    foreach ($list as $value) {
-        $adapters[] = $pools
-            ->get($value)
-            ->pop()
-            ->getResource();
-    }
-
-    return new Cache(new Sharding($adapters));
-}
-
-/**
- * Get Project DB
- *
- * @param Document $project
- * @returns Database
- */
-function getProjectDB(Document $project): Database
-{
-    global $register;
-
-    /** @var \Utopia\Pools\Group $pools */
-    $pools = $register->get('pools');
-
-    if ($project->isEmpty() || $project->getId() === 'console') {
-        return getConsoleDB();
-    }
-
-    $dbAdapter = $pools
-        ->get($project->getAttribute('database'))
-        ->pop()
-        ->getResource()
-    ;
-
-    $database = new Database($dbAdapter, getCache());
-    $database->setNamespace('_' . $project->getInternalId());
-
-    return $database;
-}
-
-
 /**
  * @param Document $database
  * @param Document $collection

src/Appwrite/Event/Audit.php

@@ -2,7 +2,6 @@
 
 namespace Appwrite\Event;
 
-use Resque;
 use Utopia\Queue\Client;
 use Utopia\Queue\Connection;
 
@@ -15,7 +14,7 @@ class Audit extends Event
 
     public function __construct(protected Connection $connection)
     {
-        parent::__construct(Event::AUDITS_QUEUE_NAME, Event::AUDITS_CLASS_NAME);
+        parent::__construct(Event::AUDITS_QUEUE_NAME, Event::AUDITS_CLASS_NAME, $connection);
     }
 
     /**

src/Appwrite/Event/Build.php

@@ -2,8 +2,9 @@
 
 namespace Appwrite\Event;
 
-use Resque;
 use Utopia\Database\Document;
+use Utopia\Queue\Client;
+use Utopia\Queue\Connection;
 
 class Build extends Event
 {
@@ -11,9 +12,9 @@ class Build extends Event
     protected ?Document $resource = null;
     protected ?Document $deployment = null;
 
-    public function __construct()
+    public function __construct(protected Connection $connection)
     {
-        parent::__construct(Event::BUILDS_QUEUE_NAME, Event::BUILDS_CLASS_NAME);
+        parent::__construct(Event::BUILDS_QUEUE_NAME, Event::BUILDS_CLASS_NAME, $connection);
     }
 
     /**
@@ -93,7 +94,11 @@ class Build extends Event
      */
     public function trigger(): string|bool
     {
-        return Resque::enqueue($this->queue, $this->class, [
+        $client = new Client($this->queue, $this->connection);
+
+        $events = $this->getEvent() ? Event::generateEvents($this->getEvent(), $this->getParams()) : null;
+
+        return $client->enqueue([
             'project' => $this->project,
             'resource' => $this->resource,
             'deployment' => $this->deployment,

src/Appwrite/Event/Certificate.php

@@ -2,7 +2,6 @@
 
 namespace Appwrite\Event;
 
-use Resque;
 use Utopia\Database\Document;
 use Utopia\Queue\Client;
 use Utopia\Queue\Connection;
@@ -14,7 +13,7 @@ class Certificate extends Event
 
     public function __construct(protected Connection $connection)
     {
-        parent::__construct(Event::CERTIFICATES_QUEUE_NAME, Event::CERTIFICATES_CLASS_NAME);
+        parent::__construct(Event::CERTIFICATES_QUEUE_NAME, Event::CERTIFICATES_CLASS_NAME, $connection);
     }
 
     /**

src/Appwrite/Event/Database.php

@@ -2,7 +2,6 @@
 
 namespace Appwrite\Event;
 
-use Resque;
 use Utopia\Database\Document;
 use Utopia\Queue\Client;
 use Utopia\Queue\Connection;
@@ -16,7 +15,7 @@ class Database extends Event
 
     public function __construct(protected Connection $connection)
     {
-        parent::__construct(Event::DATABASE_QUEUE_NAME, Event::DATABASE_CLASS_NAME);
+        parent::__construct(Event::DATABASE_QUEUE_NAME, Event::DATABASE_CLASS_NAME, $connection);
     }
 
     /**

src/Appwrite/Event/Delete.php

@@ -2,8 +2,8 @@
 
 namespace Appwrite\Event;
 
-use Resque;
 use Utopia\Database\Document;
+use Utopia\Queue\Client;
 use Utopia\Queue\Connection;
 
 class Delete extends Event
@@ -122,7 +122,11 @@ class Delete extends Event
      */
     public function trigger(): string|bool
     {
-        return Resque::enqueue($this->queue, $this->class, [
+        $client = new Client($this->queue, $this->connection);
+
+        $events = $this->getEvent() ? Event::generateEvents($this->getEvent(), $this->getParams()) : null;
+
+        return $client->enqueue([
             'project' => $this->project,
             'type' => $this->type,
             'document' => $this->document,

src/Appwrite/Event/Mail.php

@@ -2,7 +2,6 @@
 
 namespace Appwrite\Event;
 
-use Resque;
 use Utopia\Database\Document;
 use Utopia\Queue\Client;
 use Utopia\Queue\Connection;
@@ -18,7 +17,7 @@ class Mail extends Event
 
     public function __construct(protected Connection $connection)
     {
-        parent::__construct(Event::MAILS_QUEUE_NAME, Event::MAILS_CLASS_NAME);
+        parent::__construct(Event::MAILS_QUEUE_NAME, Event::MAILS_CLASS_NAME, $connection);
     }
 
     /**

src/Appwrite/Event/Phone.php

@@ -12,7 +12,7 @@ class Phone extends Event
 
     public function __construct(protected Connection $connection)
     {
-        parent::__construct(Event::MESSAGING_QUEUE_NAME, Event::MESSAGING_CLASS_NAME);
+        parent::__construct(Event::MESSAGING_QUEUE_NAME, Event::MESSAGING_CLASS_NAME, $connection);
     }
 
     /**