refactor(workers): create abstract worker class

app/controllers/api/health.php

@@ -6,7 +6,6 @@ use Utopia\Storage\Device\Local;
 use Utopia\Storage\Storage;
 use Appwrite\ClamAV\Network;
 use Appwrite\Event\Event;
-use RuntimeException;
 
 App::get('/v1/health')
     ->desc('Get HTTP')

app/workers/audits.php

@@ -1,5 +1,6 @@
 <?php
 
+use Appwrite\Resque\Worker;
 use Utopia\Audit\Audit;
 use Utopia\Audit\Adapters\MySQL as AuditAdapter;
 use Utopia\CLI\Console;
@@ -7,18 +8,17 @@ use Utopia\CLI\Console;
 require_once __DIR__.'/../init.php';
 
 Console::title('Audits V1 Worker');
-
 Console::success(APP_NAME.' audits worker v1 has started');
 
-class AuditsV1
+class AuditsV1 extends Worker
 {
     public $args = [];
 
-    public function setUp(): void
+    public function init(): void
     {
     }
 
-    public function perform()
+    public function run(): void
     {
         global $register;
 
@@ -39,8 +39,8 @@ class AuditsV1
         $audit->log($userId, $event, $resource, $userAgent, $ip, '', $data);
     }
 
-    public function tearDown(): void
+    public function shutdown(): void
     {
         // ... Remove environment for this job
     }
-}
+}

app/workers/certificates.php

@@ -1,30 +1,30 @@
 <?php
 
-use Utopia\App;
-use Utopia\CLI\Console;
-use Utopia\Config\Config;
-use Utopia\Domains\Domain;
 use Appwrite\Database\Database;
 use Appwrite\Database\Adapter\MySQL as MySQLAdapter;
 use Appwrite\Database\Adapter\Redis as RedisAdapter;
 use Appwrite\Database\Validator\Authorization;
 use Appwrite\Network\Validator\CNAME;
+use Appwrite\Resque\Worker;
+use Utopia\App;
+use Utopia\CLI\Console;
+use Utopia\Config\Config;
+use Utopia\Domains\Domain;
 
 require_once __DIR__.'/../init.php';
 
 Console::title('Certificates V1 Worker');
-
 Console::success(APP_NAME.' certificates worker v1 has started');
 
-class CertificatesV1
+class CertificatesV1 extends Worker
 {
     public $args = [];
 
-    public function setUp(): void
+    public function init(): void
     {
     }
 
-    public function perform()
+    public function run(): void
     {
         global $register;
 
@@ -204,8 +204,7 @@ class CertificatesV1
         Authorization::reset();
     }
 
-    public function tearDown(): void
+    public function shutdown(): void
     {
-        // ... Remove environment for this job
     }
-}
+}

app/workers/deletes.php

@@ -5,6 +5,7 @@ use Appwrite\Database\Adapter\MySQL as MySQLAdapter;
 use Appwrite\Database\Adapter\Redis as RedisAdapter;
 use Appwrite\Database\Document;
 use Appwrite\Database\Validator\Authorization;
+use Appwrite\Resque\Worker;
 use Utopia\Storage\Device\Local;
 use Utopia\Abuse\Abuse;
 use Utopia\Abuse\Adapters\TimeLimit;
@@ -16,22 +17,20 @@ use Utopia\Audit\Adapters\MySQL as AuditAdapter;
 require_once __DIR__.'/../init.php';
 
 Console::title('Deletes V1 Worker');
-
 Console::success(APP_NAME.' deletes worker v1 has started'."\n");
 
-class DeletesV1
+class DeletesV1 extends Worker
 {
-
     public $args = [];
 
     protected $consoleDB = null;
 
-    public function setUp(): void
+    public function init(): void
     {
     }
 
-    public function perform()
+    public function run(): void
-    { 
+    {
         $projectId = isset($this->args['projectId']) ? $this->args['projectId'] : '';
         $type = $this->args['type'];
         
@@ -82,9 +81,8 @@ class DeletesV1
             }
     }
 
-    public function tearDown(): void
+    public function shutdown(): void
     {
-        // ... Remove environment for this job
     }
     
     protected function deleteDocuments(Document $document, $projectId) 

app/workers/functions.php

@@ -6,6 +6,7 @@ use Appwrite\Database\Adapter\MySQL as MySQLAdapter;
 use Appwrite\Database\Adapter\Redis as RedisAdapter;
 use Appwrite\Database\Validator\Authorization;
 use Appwrite\Event\Event;
+use Appwrite\Resque\Worker;
 use Cron\CronExpression;
 use Swoole\Runtime;
 use Utopia\App;
@@ -125,17 +126,17 @@ Console::info(count($list)." functions listed in " . ($executionEnd - $execution
 
 //TODO aviod scheduled execution if delay is bigger than X offest
 
-class FunctionsV1
+class FunctionsV1 extends Worker
 {
     public $args = [];
 
     public $allowed = [];
 
-    public function setUp(): void
+    public function init(): void
     {
     }
 
-    public function perform()
+    public function run(): void
     {
         global $register;
 
@@ -579,7 +580,7 @@ class FunctionsV1
         return $output;
     }
 
-    public function tearDown(): void
+    public function shutdown(): void
     {
     }
-}
+}

app/workers/mails.php

@@ -1,26 +1,26 @@
 <?php
 
+use Appwrite\Resque\Worker;
 use Utopia\App;
 use Utopia\CLI\Console;
 
 require_once __DIR__.'/../init.php';
 
 Console::title('Mails V1 Worker');
-
 Console::success(APP_NAME.' mails worker v1 has started'."\n");
 
-class MailsV1
+class MailsV1 extends Worker
 {
     /**
      * @var array
      */
     public $args = [];
 
-    public function setUp(): void
+    public function init(): void
     {
     }
 
-    public function perform()
+    public function run(): void
     {
         global $register;
 
@@ -68,8 +68,7 @@ class MailsV1
         }
     }
 
-    public function tearDown(): void
+    public function shutdown(): void
     {
-        // ... Remove environment for this job
     }
-}
+}

app/workers/tasks.php

@@ -1,32 +1,32 @@
 <?php
 
-use Utopia\App;
-use Utopia\CLI\Console;
-use Utopia\Config\Config;
 use Appwrite\Database\Database;
 use Appwrite\Database\Adapter\MySQL as MySQLAdapter;
 use Appwrite\Database\Adapter\Redis as RedisAdapter;
 use Appwrite\Database\Validator\Authorization;
+use Appwrite\Resque\Worker;
 use Cron\CronExpression;
+use Utopia\App;
+use Utopia\CLI\Console;
+use Utopia\Config\Config;
 
 require_once __DIR__.'/../init.php';
 
 Console::title('Tasks V1 Worker');
-
 Console::success(APP_NAME.' tasks worker v1 has started');
 
-class TasksV1
+class TasksV1 extends Worker
 {
     /**
      * @var array
      */
     public $args = [];
 
-    public function setUp(): void
+    public function init(): void
     {
     }
 
-    public function perform()
+    public function run(): void
     {
         global $register;
 
@@ -73,11 +73,11 @@ class TasksV1
         }
 
         if ($task->getAttribute('updated') !== $updated) { // Task have already been rescheduled by owner
-            return false;
+            return;
         }
 
         if ($task->getAttribute('status') !== 'play') { // Skip task and don't schedule again
-            return false;
+            return;
         }
 
         // Reschedule
@@ -202,11 +202,10 @@ class TasksV1
 
         // Send alert if needed (use SMTP as default for now)
 
-        return true;
+        return;
     }
 
-    public function tearDown(): void
+    public function shutdown(): void
     {
-        // ... Remove environment for this job
     }
-}
+}

app/workers/usage.php

@@ -1,29 +1,30 @@
 <?php
 
+use Appwrite\Resque\Worker;
 use Utopia\App;
 use Utopia\CLI\Console;
 
 require_once __DIR__.'/../init.php';
 
 Console::title('Usage V1 Worker');
-
 Console::success(APP_NAME.' usage worker v1 has started');
 
-class UsageV1
+class UsageV1 extends Worker
 {
     /**
      * @var array
      */
     public $args = [];
 
-    public function setUp(): void
+    public function init(): void
     {
     }
 
-    public function perform()
+    public function run(): void
     {
         global $register;
 
+        /** @var \Domnikl\Statsd\Client $statsd */
         $statsd = $register->get('statsd', true);
 
         $projectId = $this->args['projectId'] ?? '';
@@ -36,12 +37,12 @@ class UsageV1
         $httpMethod = $this->args['httpMethod'] ?? '';
         $httpRequest = $this->args['httpRequest'] ?? 0;
 
-        $functionId = $this->args['functionId'];
+        $functionId = $this->args['functionId'] ?? '';
         $functionExecution = $this->args['functionExecution'] ?? 0;
         $functionExecutionTime = $this->args['functionExecutionTime'] ?? 0;
         $functionStatus = $this->args['functionStatus'] ?? '';
 
-        $tags = ",project={$projectId},version=".App::getEnv('_APP_VERSION', 'UNKNOWN').'';
+        $tags = ",project={$projectId},version=".App::getEnv('_APP_VERSION', 'UNKNOWN');
 
         // the global namespace is prepended to every key (optional)
         $statsd->setNamespace('appwrite.usage');
@@ -52,7 +53,6 @@ class UsageV1
         
         if($functionExecution >= 1) {
             $statsd->increment('executions.all'.$tags.',functionId='.$functionId.',functionStatus='.$functionStatus);
-            var_dump($tags.',functionId='.$functionId.',functionStatus='.$functionStatus);
             $statsd->count('executions.time'.$tags.',functionId='.$functionId, $functionExecutionTime);
         }
 
@@ -65,8 +65,7 @@ class UsageV1
         }
     }
 
-    public function tearDown(): void
+    public function shutdown(): void
     {
-        // ... Remove environment for this job
     }
-}
+}

app/workers/webhooks.php

@@ -1,23 +1,23 @@
 <?php
 
+use Appwrite\Resque\Worker;
 use Utopia\App;
 use Utopia\CLI\Console;
 
 require_once __DIR__.'/../init.php';
 
 Console::title('Webhooks V1 Worker');
-
 Console::success(APP_NAME.' webhooks worker v1 has started');
 
-class WebhooksV1
+class WebhooksV1 extends Worker
 {
     public $args = [];
 
-    public function setUp(): void
+    public function init(): void
     {
     }
 
-    public function perform()
+    public function run(): void
     {
         $errors = [];
 
@@ -88,8 +88,7 @@ class WebhooksV1
         }
     }
 
-    public function tearDown(): void
+    public function shutdown(): void
     {
-        // ... Remove environment for this job
     }
-}
+}

src/Appwrite/Resque/Worker.php

@@ -0,0 +1,33 @@
+<?php
+
+namespace Appwrite\Resque;
+
+use Swoole\Runtime;
+
+use function Swoole\Coroutine\run;
+
+abstract class Worker
+{
+    public $args = [];
+
+    abstract public function init(): void;
+
+    abstract public function run(): void;
+
+    abstract public function shutdown(): void;
+
+    public function setUp(): void
+    {
+        $this->init();
+    }
+
+    public function perform()
+    {
+        $this->run();
+    }
+
+    public function tearDown(): void
+    {
+        $this->shutdown();
+    }
+}