Port Certificates to new Queue system

app/cli.php

@@ -3,6 +3,7 @@
 require_once __DIR__ . '/init.php';
 require_once __DIR__ . '/controllers/general.php';
 
+use Appwrite\Event\Certificate;
 use Appwrite\Event\Func;
 use Appwrite\Platform\Appwrite;
 use Utopia\CLI\CLI;
@@ -144,6 +145,11 @@ CLI::setResource('queueForFunctions', function (Group $pools) {
     return new Func($pools->get('queue')->pop()->getResource());
 }, ['pools']);
 
+CLI::setResource('queueForCertificates', function (Group $pools) {
+    var_dump(json_encode($pools));
+    return new Certificate($pools->get('queue')->pop()->getResource());
+}, ['pools']);
+
 CLI::setResource('logError', function (Registry $register) {
     return function (Throwable $error, string $namespace, string $action) use ($register) {
         $logger = $register->get('logger');

app/controllers/api/projects.php

@@ -1314,7 +1314,8 @@ App::patch('/v1/projects/:projectId/domains/:domainId/verification')
     ->param('domainId', '', new UID(), 'Domain unique ID.')
     ->inject('response')
     ->inject('dbForConsole')
-    ->action(function (string $projectId, string $domainId, Response $response, Database $dbForConsole) {
+    ->inject('queueForCertificates')
+    ->action(function (string $projectId, string $domainId, Response $response, Database $dbForConsole, Certificate $queueForCertificates) {
 
         $project = $dbForConsole->getDocument('projects', $projectId);
 
@@ -1352,8 +1353,7 @@ App::patch('/v1/projects/:projectId/domains/:domainId/verification')
         $dbForConsole->deleteCachedDocument('projects', $project->getId());
 
         // Issue a TLS certificate when domain is verified
-        $event = new Certificate();
+        $queueForCertificates
-        $event
             ->setDomain($domain)
             ->trigger();
 

app/controllers/general.php

@@ -51,7 +51,8 @@ App::init()
     ->inject('locale')
     ->inject('clients')
     ->inject('servers')
-    ->action(function (App $utopia, Request $request, Response $response, Document $console, Document $project, Database $dbForConsole, Document $user, Locale $locale, array $clients, array $servers) {
+    ->inject('queueForCertificates')
+    ->action(function (App $utopia, Request $request, Response $response, Document $console, Document $project, Database $dbForConsole, Document $user, Locale $locale, array $clients, array $servers, Certificate $queueForCertificates) {
         /*
         * Request format
         */
@@ -122,7 +123,7 @@ App::init()
 
                         Console::info('Issuing a TLS certificate for the main domain (' . $domain->get() . ') in a few seconds...');
 
-                        (new Certificate())
+                        $queueForCertificates
                             ->setDomain($domainDocument)
                             ->trigger();
                     }

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\Certificate;
 use Appwrite\Event\Func;
 use MaxMind\Db\Reader;
 use PHPMailer\PHPMailer\PHPMailer;
@@ -860,6 +861,9 @@ App::setResource('queue', function (Group $pools) {
 App::setResource('queueForFunctions', function (Connection $queue) {
     return new Func($queue);
 }, ['queue']);
+App::setResource('queueForCertificates', function (Connection $queue) {
+    return new Certificate($queue);
+}, ['queue']);
 App::setResource('usage', function ($register) {
     return new Stats($register->get('statsd'));
 }, ['register']);

app/worker.php

@@ -2,6 +2,7 @@
 
 require_once __DIR__ . '/init.php';
 
+use Appwrite\Event\Certificate;
 use Appwrite\Event\Func;
 use Swoole\Runtime;
 use Utopia\App;
@@ -85,6 +86,16 @@ Server::setResource('queueForFunctions', function (Registry $register) {
     );
 }, ['register']);
 
+Server::setResource('queueForCertificates', function (Registry $register) {
+    $pools = $register->get('pools');
+    return new Certificate(
+        $pools
+            ->get('queue')
+            ->pop()
+            ->getResource()
+    );
+}, ['register']);
+
 Server::setResource('logger', function ($register) {
     return $register->get('logger');
 }, ['register']);

app/workers/certificates.php

@@ -1,42 +1,33 @@
 <?php
 
+require_once __DIR__ . '/../worker.php';
+
 use Appwrite\Event\Mail;
 use Appwrite\Network\Validator\CNAME;
-use Appwrite\Resque\Worker;
 use Utopia\App;
 use Utopia\CLI\Console;
+use Utopia\Database\Validator\Authorization;
 use Utopia\Database\Database;
 use Utopia\Database\Document;
-use Utopia\Database\DateTime;
 use Utopia\Database\ID;
 use Utopia\Database\Query;
 use Utopia\Domains\Domain;
+use Utopia\Queue\Message;
+use Utopia\Queue\Server;
+use Utopia\Database\DateTime;
 
-require_once __DIR__ . '/../init.php';
+Authorization::disable();
+Authorization::setDefaultStatus(false);
 
-Console::title('Certificates V1 Worker');
+$database = null;
-Console::success(APP_NAME . ' certificates worker v1 has started');
 
-class CertificatesV1 extends Worker
+Server::setResource('execute', function () {
-{
+    return function (
-    /**
+        Database $dbForConsole,
-     * Database connection shared across all methods of this file
+        Document $document,
-     *
+        Domain $domain,
-     * @var Database
+        bool $skipRenewCheck = false
-     */
+    ) {
-    private Database $dbForConsole;
-
-    public function getName(): string
-    {
-        return "certificates";
-    }
-
-    public function init(): void
-    {
-    }
-
-    public function run(): void
-    {
         /**
          * 1. Read arguments and validate domain
          * 2. Get main domain
@@ -67,14 +58,8 @@ class CertificatesV1 extends Worker
          * Note: Renewals are checked and scheduled from maintenence worker
          */
 
-        $this->dbForConsole = $this->getConsoleDB();
-
-        $skipCheck = $this->args['skipRenewCheck'] ?? false; // If true, we won't double-check expiry from cert file
-        $document = new Document($this->args['domain'] ?? []);
-        $domain = new Domain($document->getAttribute('domain', ''));
-
         // Get current certificate
-        $certificate = $this->dbForConsole->findOne('certificates', [Query::equal('domain', [$domain->get()])]);
+        $certificate = $dbForConsole->findOne('certificates', [Query::equal('domain', [$domain->get()])]);
 
         // If we don't have certificate for domain yet, let's create new document. At the end we save it
         if (!$certificate) {
@@ -90,14 +75,14 @@ class CertificatesV1 extends Worker
             }
 
             // Validate domain and DNS records. Skip if job is forced
-            if (!$skipCheck) {
+            if (!$skipRenewCheck) {
-                $mainDomain = $this->getMainDomain();
+                $mainDomain = getMainDomain($dbForConsole);
                 $isMainDomain = !isset($mainDomain) || $domain->get() === $mainDomain;
-                $this->validateDomain($domain, $isMainDomain);
+                validateDomain($domain, $isMainDomain);
             }
 
             // If certificate exists already, double-check expiry date. Skip if job is forced
-            if (!$skipCheck && !$this->isRenewRequired($domain->get())) {
+            if (!$skipRenewCheck && !isRenewRequired($domain->get())) {
                 throw new Exception('Renew isn\'t required.');
             }
 
@@ -105,7 +90,7 @@ class CertificatesV1 extends Worker
             $folder = ID::unique();
 
             // Generate certificate files using Let's Encrypt
-            $letsEncryptData = $this->issueCertificate($folder, $domain->get(), $email);
+            $letsEncryptData = issueCertificate($folder, $domain->get(), $email);
 
             // Command succeeded, store all data into document
             // We store stderr too, because it may include warnings
@@ -115,10 +100,10 @@ class CertificatesV1 extends Worker
             ]));
 
             // Give certificates to Traefik
-            $this->applyCertificateFiles($folder, $domain->get(), $letsEncryptData);
+            applyCertificateFiles($folder, $domain->get(), $letsEncryptData);
 
             // Update certificate info stored in database
-            $certificate->setAttribute('renewDate', $this->getRenewDate($domain->get()));
+            $certificate->setAttribute('renewDate', getRenewDate($domain->get()));
             $certificate->setAttribute('attempts', 0);
             $certificate->setAttribute('issueDate', DateTime::now());
         } catch (Throwable $e) {
@@ -133,43 +118,42 @@ class CertificatesV1 extends Worker
             $certificate->setAttribute('renewDate', DateTime::now());
 
             // Send email to security email
-            $this->notifyError($domain->get(), $e->getMessage(), $attempts);
+            notifyError($domain->get(), $e->getMessage(), $attempts);
         } finally {
             // All actions result in new updatedAt date
             $certificate->setAttribute('updated', DateTime::now());
 
             // Save all changes we made to certificate document into database
-            $this->saveCertificateDocument($domain->get(), $certificate);
+            saveCertificateDocument($domain->get(), $certificate, $dbForConsole);
-        }
         }
+    };
+});
 
-    public function shutdown(): void
-    {
-    }
 
 /**
  * Save certificate data into database.
  *
  * @param string $domain Domain name that certificate is for
  * @param Document $certificate Certificate document that we need to save
+ * @param Database $dbForConsole Database connection for console
  *
  * @return void
  */
-    private function saveCertificateDocument(string $domain, Document $certificate): void
+function saveCertificateDocument(string $domain, Document $certificate, Database $dbForConsole) : void
 {
     // Check if update or insert required
-        $certificateDocument = $this->dbForConsole->findOne('certificates', [Query::equal('domain', [$domain])]);
+    $certificateDocument = $dbForConsole->findOne('certificates', [Query::equal('domain', [$domain])]);
     if (!empty($certificateDocument) && !$certificateDocument->isEmpty()) {
         // Merge new data with current data
         $certificate = new Document(\array_merge($certificateDocument->getArrayCopy(), $certificate->getArrayCopy()));
 
-            $certificate = $this->dbForConsole->updateDocument('certificates', $certificate->getId(), $certificate);
+        $certificate = $dbForConsole->updateDocument('certificates', $certificate->getId(), $certificate);
     } else {
-            $certificate = $this->dbForConsole->createDocument('certificates', $certificate);
+        $certificate = $dbForConsole->createDocument('certificates', $certificate);
     }
 
     $certificateId = $certificate->getId();
-        $this->updateDomainDocuments($certificateId, $domain);
+    updateDomainDocuments($certificateId, $domain, $dbForConsole);
 }
 
 /**
@@ -177,13 +161,13 @@ class CertificatesV1 extends Worker
  *
  * @return null|string Returns main domain. If null, there is no main domain yet.
  */
-    private function getMainDomain(): ?string
+function getMainDomain($dbForConsole): ?string
 {
     $envDomain = App::getEnv('_APP_DOMAIN', '');
     if (!empty($envDomain) && $envDomain !== 'localhost') {
         return $envDomain;
     } else {
-            $domainDocument = $this->dbForConsole->findOne('domains', [Query::orderAsc('_id')]);
+        $domainDocument = $dbForConsole->findOne('domains', [Query::orderAsc('_id')]);
         if ($domainDocument) {
             return $domainDocument->getAttribute('domain');
         }
@@ -202,7 +186,7 @@ class CertificatesV1 extends Worker
  *
  * @return void
  */
-    private function validateDomain(Domain $domain, bool $isMainDomain): void
+function validateDomain(Domain $domain, bool $isMainDomain): void
 {
     if (empty($domain->get())) {
         throw new Exception('Missing certificate domain.');
@@ -214,7 +198,6 @@ class CertificatesV1 extends Worker
 
     if (!$isMainDomain) {
         // TODO: Would be awesome to also support A/AAAA records here. Maybe dry run?
-
         // Validate if domain target is properly configured
         $target = new Domain(App::getEnv('_APP_DOMAIN_TARGET', ''));
 
@@ -240,7 +223,7 @@ class CertificatesV1 extends Worker
  *
  * @return bool True, if certificate needs to be renewed
  */
-    private function isRenewRequired(string $domain): bool
+function isRenewRequired(string $domain): bool
 {
     $certPath = APP_STORAGE_CERTIFICATES . '/' . $domain . '/cert.pem';
     if (\file_exists($certPath)) {
@@ -271,7 +254,7 @@ class CertificatesV1 extends Worker
  *
  * @return array Named array with keys 'stdout' and 'stderr', both string
  */
-    private function issueCertificate(string $folder, string $domain, string $email): array
+function issueCertificate(string $folder, string $domain, string $email): array
 {
     $stdout = '';
     $stderr = '';
@@ -301,7 +284,7 @@ class CertificatesV1 extends Worker
  *
  * @return string
  */
-    private function getRenewDate(string $domain): string
+function getRenewDate(string $domain): string
 {
     $certPath = APP_STORAGE_CERTIFICATES . '/' . $domain . '/cert.pem';
     $certData = openssl_x509_parse(file_get_contents($certPath));
@@ -319,7 +302,7 @@ class CertificatesV1 extends Worker
  *
  * @return void
  */
-    private function applyCertificateFiles(string $folder, string $domain, array $letsEncryptData): void
+function applyCertificateFiles(string $folder, string $domain, array $letsEncryptData): void
 {
     // Prepare folder in storage for domain
     $path = APP_STORAGE_CERTIFICATES . '/' . $domain;
@@ -368,7 +351,7 @@ class CertificatesV1 extends Worker
  *
  * @return void
  */
-    private function notifyError(string $domain, string $errorMessage, int $attempt): void
+function notifyError(string $domain, string $errorMessage, int $attempt): void
 {
     // Log error into console
     Console::warning('Cannot renew domain (' . $domain . ') on attempt no. ' . $attempt . ' certificate: ' . $errorMessage);
@@ -398,12 +381,13 @@ class CertificatesV1 extends Worker
  *
  * @param string $certificateId ID of a new or updated certificate document
  * @param string $domain Domain that is affected by new certificate
+ * @param Database $dbForConsole Database instance for console
  *
  * @return void
  */
-    private function updateDomainDocuments(string $certificateId, string $domain): void
+function updateDomainDocuments(string $certificateId, string $domain, Database $dbForConsole): void
 {
-        $domains = $this->dbForConsole->find('domains', [
+    $domains = $dbForConsole->find('domains', [
         Query::equal('domain', [$domain]),
         Query::limit(1000),
     ]);
@@ -412,11 +396,31 @@ class CertificatesV1 extends Worker
         $domainDocument->setAttribute('updated', DateTime::now());
         $domainDocument->setAttribute('certificateId', $certificateId);
 
-            $this->dbForConsole->updateDocument('domains', $domainDocument->getId(), $domainDocument);
+        $dbForConsole->updateDocument('domains', $domainDocument->getId(), $domainDocument);
 
         if ($domainDocument->getAttribute('projectId')) {
-                $this->dbForConsole->deleteCachedDocument('projects', $domainDocument->getAttribute('projectId'));
+            $dbForConsole->deleteCachedDocument('projects', $domainDocument->getAttribute('projectId'));
         }
     }
 }
+
+$server->job()
+    ->inject('message')
+    ->inject('dbForConsole')
+    ->inject('execute')
+    ->action(function ($message, $dbForConsole, $execute) use ($server) {
+        $payload = $message->getPayload() ?? [];
+
+        if (empty($payload)) {
+            throw new Exception('Missing payload');
         }
+
+        $document = new Document($payload['domain'] ?? []);
+        $domain = new Domain($document->getAttribute('domain', ''));
+        $skipRenewCheck = $payload['skipRenewCheck'] ?? false;
+
+        $execute($dbForConsole, $document, $domain, $skipRenewCheck);
+    });
+
+$server->workerStart();
+$server->start();

docker-compose.yml

@@ -589,6 +589,7 @@ services:
       - _APP_MAINTENANCE_RETENTION_AUDIT
       - _APP_MAINTENANCE_RETENTION_USAGE_HOURLY
       - _APP_MAINTENANCE_RETENTION_SCHEDULES
+      - _APP_CONNECTIONS_QUEUE
 
   appwrite-usage:
     entrypoint: usage

src/Appwrite/Event/Certificate.php

@@ -4,13 +4,15 @@ namespace Appwrite\Event;
 
 use Resque;
 use Utopia\Database\Document;
+use Utopia\Queue\Client;
+use Utopia\Queue\Connection;
 
 class Certificate extends Event
 {
     protected bool $skipRenewCheck = false;
     protected ?Document $domain = null;
 
-    public function __construct()
+    public function __construct(protected Connection $connection)
     {
         parent::__construct(Event::CERTIFICATES_QUEUE_NAME, Event::CERTIFICATES_CLASS_NAME);
     }
@@ -69,7 +71,9 @@ class Certificate extends Event
      */
     public function trigger(): string|bool
     {
-        return Resque::enqueue($this->queue, $this->class, [
+        $client = new Client($this->queue, $this->connection);
+
+        return $client->enqueue([
             'project' => $this->project,
             'domain' => $this->domain,
             'skipRenewCheck' => $this->skipRenewCheck

src/Appwrite/Platform/Tasks/Maintenance.php

@@ -25,10 +25,11 @@ class Maintenance extends Action
         $this
             ->desc('Schedules maintenance tasks and publishes them to resque')
             ->inject('dbForConsole')
-            ->callback(fn (Database $dbForConsole) => $this->action($dbForConsole));
+            ->inject('queueForCertificates')
+            ->callback(fn (Database $dbForConsole, Certificate $queueForCertificates) => $this->action($dbForConsole, $queueForCertificates));
     }
 
-    public function action(Database $dbForConsole): void
+    public function action(Database $dbForConsole, Certificate $queueForCertificates): void
     {
         Console::title('Maintenance V1');
         Console::success(APP_NAME . ' maintenance process v1 has started');
@@ -80,7 +81,7 @@ class Maintenance extends Action
                 ->trigger();
         }
 
-        function renewCertificates($dbForConsole)
+        function renewCertificates($dbForConsole, $queueForCertificates)
         {
             $time = DateTime::now();
 
@@ -91,12 +92,11 @@ class Maintenance extends Action
             ]);
 
 
-            if (\count($certificates) > 0) {
+            if (\count($certificates) > 0 || true) {
                 Console::info("[{$time}] Found " . \count($certificates) . " certificates for renewal, scheduling jobs.");
 
-                $event = new Certificate();
                 foreach ($certificates as $certificate) {
-                    $event
+                    $queueForCertificates
                         ->setDomain(new Document([
                             'domain' => $certificate->getAttribute('domain')
                         ]))
@@ -135,7 +135,7 @@ class Maintenance extends Action
         $cacheRetention = (int) App::getEnv('_APP_MAINTENANCE_RETENTION_CACHE', '2592000'); // 30 days
         $schedulesDeletionRetention = (int) App::getEnv('_APP_MAINTENANCE_RETENTION_SCHEDULES', '86400'); // 1 Day
 
-        Console::loop(function () use ($interval, $executionLogsRetention, $abuseLogsRetention, $auditLogRetention, $cacheRetention, $schedulesDeletionRetention, $usageStatsRetentionHourly, $dbForConsole) {
+        Console::loop(function () use ($interval, $executionLogsRetention, $abuseLogsRetention, $auditLogRetention, $cacheRetention, $schedulesDeletionRetention, $usageStatsRetentionHourly, $dbForConsole, $queueForCertificates) {
             $time = DateTime::now();
 
             Console::info("[{$time}] Notifying workers with maintenance tasks every {$interval} seconds");
@@ -145,7 +145,7 @@ class Maintenance extends Action
             notifyDeleteUsageStats($usageStatsRetentionHourly);
             notifyDeleteConnections();
             notifyDeleteExpiredSessions();
-            renewCertificates($dbForConsole);
+            renewCertificates($dbForConsole, $queueForCertificates);
             notifyDeleteCache($cacheRetention);
             notifyDeleteSchedules($schedulesDeletionRetention);
         }, $interval);