Merge pull request #7497 from appwrite/feat-queue-commands

Add queue management commands

Dockerfile

@@ -85,6 +85,9 @@ RUN chmod +x /usr/local/bin/doctor && \
     chmod +x /usr/local/bin/test && \
     chmod +x /usr/local/bin/vars && \
     chmod +x /usr/local/bin/queue-retry && \
+    chmod +x /usr/local/bin/queue-count-failed && \
+    chmod +x /usr/local/bin/queue-count-processing && \
+    chmod +x /usr/local/bin/queue-count-success && \
     chmod +x /usr/local/bin/worker-audits && \
     chmod +x /usr/local/bin/worker-certificates && \
     chmod +x /usr/local/bin/worker-databases && \

bin/queue-count-failed

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+php /usr/src/code/app/cli.php queue-count --type=failed $@

bin/queue-count-processing

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+php /usr/src/code/app/cli.php queue-count --type=processing $@

bin/queue-count-success

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+php /usr/src/code/app/cli.php queue-count --type=success $@

src/Appwrite/Platform/Services/Tasks.php

@@ -22,6 +22,7 @@ use Appwrite\Platform\Tasks\GetMigrationStats;
 use Appwrite\Platform\Tasks\PatchRecreateRepositoriesDocuments;
 use Appwrite\Platform\Tasks\QueueRetry;
 use Appwrite\Platform\Tasks\CreateInfMetric;
+use Appwrite\Platform\Tasks\QueueCount;
 
 class Tasks extends Service
 {
@@ -47,6 +48,7 @@ class Tasks extends Service
             ->addAction(PatchRecreateRepositoriesDocuments::getName(), new PatchRecreateRepositoriesDocuments())
             ->addAction(GetMigrationStats::getName(), new GetMigrationStats())
             ->addAction(QueueRetry::getName(), new QueueRetry())
+            ->addAction(QueueCount::getName(), new QueueCount())
             ->addAction(CreateInfMetric::getName(), new CreateInfMetric())
         ;
     }

src/Appwrite/Platform/Tasks/QueueCount.php

@@ -0,0 +1,77 @@
+<?php
+
+namespace Appwrite\Platform\Tasks;
+
+use Appwrite\Event\Event;
+use Utopia\CLI\Console;
+use Utopia\Platform\Action;
+use Utopia\Queue\Client;
+use Utopia\Queue\Connection;
+use Utopia\Validator\WhiteList;
+
+class QueueCount extends Action
+{
+    public static function getName(): string
+    {
+        return 'queue-count';
+    }
+
+
+    public function __construct()
+    {
+        $this
+            ->desc('Return the number of from a specific queue identified by the name parameter with a specific type')
+            ->param('name', '', new WhiteList([
+                Event::DATABASE_QUEUE_NAME,
+                Event::DELETE_QUEUE_NAME,
+                Event::AUDITS_QUEUE_NAME,
+                Event::MAILS_QUEUE_NAME,
+                Event::FUNCTIONS_QUEUE_NAME,
+                Event::USAGE_QUEUE_NAME,
+                Event::WEBHOOK_QUEUE_NAME,
+                Event::CERTIFICATES_QUEUE_NAME,
+                Event::BUILDS_QUEUE_NAME,
+                Event::MESSAGING_QUEUE_NAME,
+                Event::MIGRATIONS_QUEUE_NAME,
+                Event::HAMSTER_QUEUE_NAME
+            ]), 'Queue name')
+            ->param('type', '', new WhiteList([
+                'success',
+                'failed',
+                'processing',
+            ]), 'Queue type')
+            ->inject('queue')
+            ->callback(fn ($name, $type, $queue) => $this->action($name, $type, $queue));
+    }
+
+    /**
+     * @param string $name The name of the queue to count the jobs from
+     * @param string $type The type of jobs to count
+     * @param Connection $queue
+     */
+    public function action(string $name, string $type, Connection $queue): void
+    {
+        if (!$name) {
+            Console::error('Missing required parameter $name');
+            return;
+        }
+
+        $queueClient = new Client($name, $queue);
+
+        $count = 0;
+
+        switch ($type) {
+            case 'success':
+                $count = $queueClient->countSuccessfulJobs();
+                break;
+            case 'failed':
+                $count = $queueClient->countFailedJobs();
+                break;
+            case 'processing':
+                $count = $queueClient->countProcessingJobs();
+                break;
+        };
+
+        Console::log("Queue: '{$name}' has {$count} {$type} jobs.");
+    }
+}