Set queue name per worker

.env

@@ -1,10 +1,10 @@
 _APP_ENV=development
 _APP_LOCALE=en
 _APP_WORKER_PER_CORE=6
+_APP_CONNECTIONS_DB_QUEUE_01=database_db_main
 _APP_CONSOLE_WHITELIST_ROOT=disabled
 _APP_CONSOLE_WHITELIST_EMAILS=
 _APP_CONSOLE_WHITELIST_IPS=
-_APP_CONNECTIONS_DB_QUEUE=database-db-main
 _APP_SYSTEM_EMAIL_NAME=Appwrite
 _APP_SYSTEM_EMAIL_ADDRESS=team@appwrite.io
 _APP_SYSTEM_SECURITY_EMAIL_ADDRESS=security@appwrite.io

app/worker.php

@@ -18,13 +18,11 @@ use Swoole\Runtime;
 use Utopia\App;
 use Utopia\Cache\Adapter\Sharding;
 use Utopia\Cache\Cache;
-use Utopia\CLI\CLI;
 use Utopia\CLI\Console;
 use Utopia\Config\Config;
 use Utopia\Database\Database;
 use Utopia\Database\Document;
 use Utopia\Database\Validator\Authorization;
-use Utopia\Queue\Adapter\Swoole;
 use Utopia\Platform\Service;
 use Utopia\Queue\Message;
 use Utopia\Queue\Server;
@@ -134,9 +132,6 @@ Server::setResource('queueForMails', function (Connection $queue) {
 Server::setResource('queueForBuilds', function (Connection $queue) {
     return new Build($queue);
 }, ['queue']);
-Server::setResource('queueForDatabase', function (Connection $queue) {
-    return new EventDatabase($queue);
-}, ['queue']);
 Server::setResource('queueForDeletes', function (Connection $queue) {
     return new Delete($queue);
 }, ['queue']);
@@ -216,17 +211,25 @@ $pools = $register->get('pools');
 $platform = new Appwrite();
 $args = $_SERVER['argv'];
 
-if (isset($args[0])) {
-    $workerName = end($args);
-} else {
+if (!isset($args[1])) {
     Console::error('Missing worker name');
+    Console::exit(1);
+}
+
+\array_shift($args);
+$workerName = $args[0];
+$workerIndex = $args[1] ?? '';
+
+if (!empty($workerNum)) {
+    $workerName .= '_' . $workerIndex;
 }
 
 try {
     $platform->init(Service::TYPE_WORKER, [
-        'workersNum' => strtolower($workerName) === 'databases'? 1 :swoole_cpu_num() * intval(App::getEnv('_APP_WORKER_PER_CORE', 6)),
+        'workersNum' => str_starts_with(strtolower($workerName), 'databases') ? 1 : swoole_cpu_num() * intval(App::getEnv('_APP_WORKER_PER_CORE', 6)),
         'connection' => $pools->get('queue')->pop()->getResource(),
         'workerName' => strtolower($workerName) ?? null,
+        'queueName' => App::getEnv('_APP_CONNECTIONS_DB_QUEUE', 'database_db_main')
     ]);
 } catch (\Exception $e) {
     Console::error($e->getMessage() . ', File: ' . $e->getFile() .  ', Line: ' . $e->getLine());

composer.json

@@ -56,7 +56,7 @@
         "utopia-php/image": "0.5.*",
         "utopia-php/locale": "0.4.*",
         "utopia-php/logger": "0.3.*",
-        "utopia-php/messaging": "0.1.*",
+        "utopia-php/messaging": "0.2.*",
         "utopia-php/migration": "0.3.*",
         "utopia-php/orchestration": "0.9.*",
         "utopia-php/platform": "dev-integrate-workers as 0.3.3",
@@ -64,7 +64,7 @@
         "utopia-php/preloader": "0.2.*",
         "utopia-php/queue": "dev-feat-get-worker-start as 0.5.3",
         "utopia-php/registry": "0.5.*",
-        "utopia-php/storage": "0.14.*",
+        "utopia-php/storage": "0.17.*",
         "utopia-php/swoole": "0.5.*",
         "utopia-php/vcs": "0.5.*",
         "utopia-php/websocket": "0.1.*",

composer.lock

@@ -2007,12 +2007,12 @@
             "source": {
                 "type": "git",
                 "url": "https://github.com/utopia-php/platform.git",
-                "reference": "f7fe90764f0bcf73eea6f968965b2d786822033b"
+                "reference": "056a60bb65eb12005f451714da0fe35a22f424c2"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/utopia-php/platform/zipball/f7fe90764f0bcf73eea6f968965b2d786822033b",
-                "reference": "f7fe90764f0bcf73eea6f968965b2d786822033b",
+                "url": "https://api.github.com/repos/utopia-php/platform/zipball/056a60bb65eb12005f451714da0fe35a22f424c2",
+                "reference": "056a60bb65eb12005f451714da0fe35a22f424c2",
                 "shasum": ""
             },
             "require": {
@@ -2046,9 +2046,9 @@
             ],
             "support": {
                 "issues": "https://github.com/utopia-php/platform/issues",
-                "source": "https://github.com/utopia-php/platform/tree/integrate-workers"
+                "source": "https://github.com/utopia-php/platform/tree/feat-custom-queue-names"
             },
-            "time": "2023-10-04T14:44:45+00:00"
+            "time": "2023-10-12T04:31:27+00:00"
         },
         {
             "name": "utopia-php/pools",
@@ -5316,5 +5316,5 @@
     "platform-overrides": {
         "php": "8.0"
     },
-    "plugin-api-version": "2.2.0"
+    "plugin-api-version": "2.3.0"
 }

docker-compose.yml

@@ -356,10 +356,10 @@ services:
       - _APP_EXECUTOR_SECRET
       - _APP_EXECUTOR_HOST
 
-  appwrite-worker-databases:
-    entrypoint: worker-databases
+  appwrite-worker-databases-01:
+    entrypoint: worker-databases 01
     <<: *x-logging
-    container_name: appwrite-worker-databases
+    container_name: appwrite-worker-databases-01
     image: appwrite-dev
     networks:
       - appwrite
@@ -372,7 +372,7 @@ services:
       - mariadb
     environment:
       - _APP_ENV
-      - _APP_CONNECTIONS_DB_QUEUE
+      - _APP_CONNECTIONS_DB_QUEUE_01
       - _APP_WORKER_PER_CORE
       - _APP_OPENSSL_KEY_V1
       - _APP_REDIS_HOST

src/Appwrite/Event/Database.php

@@ -18,9 +18,7 @@ class Database extends Event
     {
         parent::__construct($connection);
 
-        $this
-            ->setQueue($this->getProject()->getAttribute('database'))
-            ->setClass(Event::DATABASE_CLASS_NAME);
+        $this->setClass(Event::DATABASE_CLASS_NAME);
     }
 
     /**
@@ -110,6 +108,8 @@ class Database extends Event
      */
     public function trigger(): string|bool
     {
+        $this->setQueue($this->getProject()->getAttribute('database'));
+
         $client = new Client($this->queue, $this->connection);
 
         return $client->enqueue([