Merge pull request #538 from arabcoders/dev

Migrated tasks to use the new events system for queue.

Author: arabcoders

FAQ.md

@@ -913,25 +913,37 @@ is an example of how to do it for debian based systems.
 ```yaml
 services:
     watchstate:
-        image: ghcr.io/arabcoders/watchstate:latest
-        # To change the user/group id associated with the tool change the following line.
-        user: "${UID:-1000}:${GID:-1000}"
-        group_add:
-            - "44" # Add video group to the container.
-            - "110" # Add render group to the container.
         container_name: watchstate
+        image: ghcr.io/arabcoders/watchstate:latest   # The image to use. you can use the latest or dev tag.
+        user: "${UID:-1000}:${GID:-1000}"             # user and group id to run the container under. 
+        group_add:
+            - "44"                                    # Add video group to the container.
+            - "105"                                   # Add render group to the container.
         restart: unless-stopped
         ports:
-            - "8080:8080" # The port which will serve WebUI + API + Webhooks
+            - "8080:8080"                             # The port which will serve WebUI + API + Webhooks
+        devices:
+            - /dev/dri:/dev/dri                       # mount the dri devices to the container.
         volumes:
-            - ./data:/config:rw # mount current directory to container /config directory.
-            - /dev/dri:/dev/dri # mount the dri devices to the container.
-            - /storage/media:/media:ro # mount your media directory to the container.
+            - ./data:/config:rw                       # mount current directory to container /config directory.
+            - /storage/media:/media:ro                # mount your media directory to the container.
 ```
 
-This setup should work for VAAPI encoding in `x86_64` containers, for other architectures you need to adjust the
-`/dev/dri` to match your hardware. There are currently an issue with nvidia h264_nvenc encoding, the alpine build for
-`ffmpeg`doesn't include the codec.
+This setup should work for VAAPI encoding in `x86_64` containers, There are currently an issue with nvidia h264_nvenc
+encoding, the alpine build for`ffmpeg` doesn't include the codec. i am looking for a way include the codec without
+ballooning the image size by 600MB+. If you have a solution please let me know.
 
-Note: the tip about adding the group_add came from the user `binarypancakes` in discord.
+Please know that your `video`, `render` group id might be different then mine, you can run the follow command in docker
+host server to get the group ids for both groups.
+
+```bash
+$ cat /etc/group | grep -E 'render|video'
 
+video:x:44:your_docker_username
+render:x:105:your_docker_username
+```
+
+In my docker host the group id for `video` is `44` and for `render` is `105`. change what needed in the `compose.yaml`
+file to match your setup.
+
+Note: the tip about adding the group_add came from the user `binarypancakes` in discord.

config/config.php

@@ -299,6 +299,7 @@
                 'enabled' => true,
                 'timer' => '* * * * *',
                 'args' => '-v',
+                'hide' => true,
             ],
         ],
     ];

container/files/init-container.sh

@@ -88,6 +88,9 @@ fi
 echo "[$(date +"%Y-%m-%dT%H:%M:%S%z")] Caching tool routes."
 /opt/bin/console system:routes
 
+echo "[$(date +"%Y-%m-%dT%H:%M:%S%z")] Caching events listeners."
+/opt/bin/console events:cache
+
 echo "[$(date +"%Y-%m-%dT%H:%M:%S%z")] Running database migrations."
 /opt/bin/console system:db:migrations
 

frontend/pages/events/index.vue

@@ -85,7 +85,12 @@
               </time>
             </span>
             <span class="card-footer-item">
-              <span v-if="!item.updated_at" class="icon"><i class="fas fa-spinner fa-spin"></i></span>
+              <template v-if="!item.updated_at">
+                <span v-if="0 === item.status" class="icon">
+                  <i class="fas fa-spinner fa-spin"></i>
+                </span>
+                <span v-else>None</span>
+              </template>
               <template v-else>
                 <span class="icon"><i class="fas fa-calendar-alt"></i></span>
                 <time class="has-tooltip" v-tooltip="`Updated at: ${moment(item.updated_at)}`">

frontend/pages/tasks.vue

@@ -122,7 +122,7 @@
                   <span class="icon"><i class="fas fa-clock" :class="{ 'fa-spin': task.queued }"></i></span>
                   <span>
                     <template v-if="!task.queued">Queue Task</template>
-                    <template v-else>Remove from queue</template>
+                    <template v-else>Cancel Task</template>
                   </span>
                 </span>
               </button>
@@ -239,7 +239,7 @@ const queueTask = async task => {
   try {
     const response = await request(`/tasks/${task.name}/queue`, {method: is_queued ? 'DELETE' : 'POST'})
     if (response.ok) {
-      notification('success', 'Success', `Task '${task.name}' has been ${is_queued ? 'removed from the queue' : 'queued'}.`)
+      notification('success', 'Success', `Task '${task.name}' has been ${is_queued ? 'cancelled' : 'queued'}.`)
       task.queued = !is_queued
       if (task.queued) {
         queued.value.push(task.name)

src/API/Tasks/Index.php renamed to src/API/Tasks.php

@@ -2,24 +2,26 @@
 
 declare(strict_types=1);
 
-namespace App\API\Tasks;
+namespace App\API;
 
 use App\Commands\System\TasksCommand;
 use App\Libs\Attributes\Route\Get;
 use App\Libs\Attributes\Route\Route;
 use App\Libs\Enums\Http\Status;
+use App\Model\Events\Event;
+use App\Model\Events\EventsRepository;
+use App\Model\Events\EventsTable;
+use App\Model\Events\EventStatus;
 use Cron\CronExpression;
-use DateInterval;
 use Psr\Http\Message\ResponseInterface as iResponse;
 use Psr\Http\Message\ServerRequestInterface as iRequest;
-use Psr\SimpleCache\CacheInterface as iCache;
 use Psr\SimpleCache\InvalidArgumentException;
 
-final class Index
+final class Tasks
 {
     public const string URL = '%{api.prefix}/tasks';
 
-    public function __construct(private readonly iCache $cache)
+    public function __construct(private EventsRepository $eventsRepo)
     {
     }
 
@@ -29,22 +31,28 @@ public function __construct(private readonly iCache $cache)
     #[Get(self::URL . '[/]', name: 'tasks.index')]
     public function tasksIndex(): iResponse
     {
-        $queuedTasks = $this->cache->get('queued_tasks', []);
-        $response = [
-            'tasks' => [],
-            'queued' => $queuedTasks,
-            'status' => isTaskWorkerRunning(),
-        ];
+        $tasks = [];
 
         foreach (TasksCommand::getTasks() as $task) {
             $task = self::formatTask($task);
-            $task['queued'] = in_array(ag($task, 'name'), $queuedTasks);
+            if (true === (bool)ag($task, 'hide', false)) {
+                continue;
+            }
 
+            $task['queued'] = null !== $this->isQueued(ag($task, 'name'));
+            $tasks[] = $task;
+        }
 
-            $response['tasks'][] = $task;
+        $queued = [];
+        foreach (array_filter($tasks, fn($item) => $item['queued'] === true) as $item) {
+            $queued[] = $item['name'];
         }
 
-        return api_response(Status::OK, $response);
+        return api_response(Status::OK, [
+            'tasks' => $tasks,
+            'queued' => $queued,
+            'status' => isTaskWorkerRunning(),
+        ]);
     }
 
     /**
@@ -59,29 +67,41 @@ public function taskQueue(iRequest $request, string $id): iResponse
             return api_error('Task not found.', Status::NOT_FOUND);
         }
 
-        $queuedTasks = $this->cache->get('queued_tasks', []);
+        $queuedTask = $this->isQueued(ag($task, 'name'));
 
         if ('POST' === $request->getMethod()) {
-            $queuedTasks[] = $id;
-            $this->cache->set('queued_tasks', $queuedTasks, new DateInterval('P3D'));
-            return api_response(Status::ACCEPTED, ['queue' => $queuedTasks]);
+            if (null !== $queuedTask) {
+                return api_error('Task already queued.', Status::CONFLICT);
+            }
+
+            $event = queueEvent(TasksCommand::NAME, ['name' => $id], [
+                EventsTable::COLUMN_REFERENCE => r('task://{name}', ['name' => $id]),
+            ]);
+
+            return api_response(Status::ACCEPTED, $event->getAll());
         }
 
         if ('DELETE' === $request->getMethod()) {
-            $queuedTasks = array_filter($queuedTasks, fn($v) => $v !== $id);
-            $this->cache->set('queued_tasks', $queuedTasks, new DateInterval('P3D'));
-            return api_response(Status::OK, ['queue' => $queuedTasks]);
+            if (null === $queuedTask) {
+                return api_error('Task not queued.', Status::NOT_FOUND);
+            }
+
+            if ($queuedTask->status === EventStatus::RUNNING) {
+                return api_error('Cannot remove task in running state.', Status::BAD_REQUEST);
+            }
+
+            $queuedTask->status = EventStatus::CANCELLED;
+            $this->eventsRepo->save($queuedTask);
+
+            return api_response(Status::OK);
         }
 
         return api_response(Status::OK, [
             'task' => $id,
-            'is_queued' => in_array($id, $queuedTasks),
+            'is_queued' => null !== $queuedTask,
         ]);
     }
 
-    /**
-     * @throws InvalidArgumentException
-     */
     #[Get(self::URL . '/{id:[a-zA-Z0-9_-]+}[/]', name: 'tasks.task.view')]
     public function taskView(string $id): iResponse
     {
@@ -91,10 +111,8 @@ public function taskView(string $id): iResponse
             return api_error('Task not found.', Status::NOT_FOUND);
         }
 
-        $queuedTasks = $this->cache->get('queued_tasks', []);
-
-        $data = Index::formatTask($task);
-        $data['queued'] = in_array(ag($task, 'name'), $queuedTasks);
+        $data = Tasks::formatTask($task);
+        $data['queued'] = null !== $this->isQueued(ag($task, 'name'));
 
         return api_response(Status::OK, $data);
     }
@@ -115,6 +133,7 @@ private function formatTask(array $task): array
             'prev_run' => null,
             'command' => ag($task, 'command'),
             'args' => ag($task, 'args'),
+            'hide' => (bool)ag($task, 'hide', false),
         ];
 
         if (!is_string($item['command'])) {
@@ -136,4 +155,11 @@ private function formatTask(array $task): array
 
         return $item;
     }
+
+    private function isQueued(string $id): Event|null
+    {
+        return $this->eventsRepo->findByReference(r('task://{name}', ['name' => $id]), [
+            EventsTable::COLUMN_STATUS => EventStatus::PENDING->value
+        ]);
+    }
 }

src/Commands/Events/DispatchCommand.php

@@ -142,7 +142,7 @@ private function runEvent(Event $event): void
             $event->updated_at = (string)makeDate();
             $this->repo->save($event);
 
-            $this->logger->error($errorLog);
+            $this->logger->error($errorLog, ['trace' => $e->getTrace()]);
         }
     }
 }

src/Commands/System/IndexCommand.php

@@ -20,9 +20,9 @@
 #[Cli(command: self::ROUTE)]
 final class IndexCommand extends Command
 {
-    public const ROUTE = 'system:index';
+    public const string ROUTE = 'system:index';
 
-    public const TASK_NAME = 'indexes';
+    public const string TASK_NAME = 'indexes';
 
     /**
      * Class constructor.