Refactor and cleanups

Author: marickvantuil

README.md

@@ -59,10 +59,6 @@ Add a new queue connection to `config/queue.php`
   'handler'               => env('STACKKIT_CLOUD_TASKS_HANDLER', ''),
   'service_account_email' => env('STACKKIT_CLOUD_TASKS_SERVICE_EMAIL', ''),
 
-  // Optional: The deadline in seconds for requests sent to the worker. If the worker
-  // does not respond by this deadline then the request is cancelled and the attempt
-  // is marked as a DEADLINE_EXCEEDED failure.
-  'dispatch_deadline' => null,
   'backoff' => 0,
 ],
 ```

assets/dashboard.png

@@ -8,9 +8,4 @@
 
     // If the application only dispatches jobs
     'disable_task_handler' => env('STACKKIT_CLOUD_TASKS_DISABLE_TASK_HANDLER', false),
-
-    'dashboard' => [
-        'enabled' => env('STACKKIT_CLOUD_TASKS_DASHBOARD_ENABLED', false),
-        'password' => env('STACKKIT_CLOUD_TASKS_DASHBOARD_PASSWORD', 'MyPassword1!'),
-    ],
 ];

config/cloud-tasks.php

@@ -62,31 +62,21 @@ public function setAttempts(int $attempts): void
         $this->job['internal']['attempts'] = $attempts;
     }
 
-    public function getTaskName(): string
-    {
-        return $this->job['internal']['taskName'];
-    }
-
     public function delete(): void
     {
         // Laravel automatically calls delete() after a job is processed successfully.
         // However, this is not what we want to happen in Cloud Tasks because Cloud Tasks
         // will also delete the task upon a 200 OK status, which means a task is deleted twice.
     }
 
-    public function hasError(): bool
-    {
-        return data_get($this->job, 'internal.errored') === true;
-    }
-
     public function release($delay = 0): void
     {
         parent::release($delay);
 
         $this->driver->release($this, $delay);
 
         if (! data_get($this->job, 'internal.errored')) {
-            app('events')->dispatch(new JobReleased($this->getConnectionName(), $this, $delay));
+            event(new JobReleased($this->getConnectionName(), $this, $delay));
         }
     }
 }

migrations/2024_02_14_211616_create_cloud_tasks_table.php

@@ -12,7 +12,6 @@
 use Google\Cloud\Tasks\V2\HttpRequest;
 use Google\Cloud\Tasks\V2\OidcToken;
 use Google\Cloud\Tasks\V2\Task;
-use Google\Protobuf\Duration;
 use Google\Protobuf\Timestamp;
 use Illuminate\Contracts\Queue\Queue as QueueContract;
 use Illuminate\Queue\Queue as LaravelQueue;
@@ -138,22 +137,10 @@ protected function pushToCloudTasks($queue, $payload, $delay, mixed $job)
 
         $task = tap(new Task())->setName($this->taskName($queue, $payload['displayName']));
 
-        $payload = $this->enrichPayloadWithInternalData(
-            payload: $payload,
-            queueName: $queue,
-            taskName: $task->getName(),
-            connectionName: $this->getConnectionName(),
-        );
+        $payload = $this->enrichPayloadWithAttempts($payload);
 
         $this->addPayloadToTask($payload, $task, $job);
 
-        // The deadline for requests sent to the app. If the app does not respond by
-        // this deadline then the request is cancelled and the attempt is marked as
-        // a failure. Cloud Tasks will retry the task according to the RetryConfig.
-        if (! empty($this->config['dispatch_deadline'])) {
-            $task->setDispatchDeadline(new Duration(['seconds' => $this->config['dispatch_deadline']]));
-        }
-
         $availableAt = $this->availableAt($delay);
         if ($availableAt > time()) {
             $task->setScheduleTime(new Timestamp(['seconds' => $availableAt]));
@@ -175,22 +162,18 @@ private function taskName(string $queueName, string $displayName): string
             $queueName,
             str($displayName)
                 ->afterLast('\\')
+                ->replaceMatches('![^-\pL\pN\s]+!u', '-')
+                ->replaceMatches('![-\s]+!u', '-')
                 ->prepend((string) Str::ulid(), '-')
                 ->toString(),
         );
     }
 
-    private function enrichPayloadWithInternalData(
+    private function enrichPayloadWithAttempts(
         array $payload,
-        string $queueName,
-        string $taskName,
-        string $connectionName,
     ): array {
         $payload['internal'] = [
             'attempts' => $payload['internal']['attempts'] ?? 0,
-            'queue' => $queueName,
-            'taskName' => $taskName,
-            'connection' => $connectionName,
         ];
 
         return $payload;
@@ -241,7 +224,7 @@ public function pop($queue = null)
 
     public function delete(CloudTasksJob $job): void
     {
-        CloudTasksApi::deleteTask($job->getTaskName());
+        // Job deletion will be handled by Cloud Tasks.
     }
 
     public function release(CloudTasksJob $job, int $delay = 0): void

src/CloudTasksJob.php

@@ -95,20 +95,6 @@ private function registerEvents(): void
             );
         });
 
-        $events->listen(JobProcessing::class, function (JobProcessing $event) {
-            if (! $event->job instanceof CloudTasksJob) {
-                return;
-            }
-        });
-
-        $events->listen(JobProcessed::class, function (JobProcessed $event) {
-            if (! $event->job instanceof CloudTasksJob) {
-                return;
-            }
-
-            data_set($event->job->job, 'internal.processed', true);
-        });
-
         $events->listen(JobExceptionOccurred::class, function (JobExceptionOccurred $event) {
             if (! $event->job instanceof CloudTasksJob) {
                 return;

src/CloudTasksQueue.php

@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Stackkit\LaravelGoogleCloudTasksQueue\Events;
+
+use Stackkit\LaravelGoogleCloudTasksQueue\IncomingTask;
+
+class TaskIncoming
+{
+    public function __construct(public IncomingTask $task)
+    {
+        //
+    }
+}

src/CloudTasksServiceProvider.php

@@ -4,6 +4,8 @@
 
 namespace Stackkit\LaravelGoogleCloudTasksQueue;
 
+use Error;
+use Illuminate\Contracts\Encryption\Encrypter;
 use Safe\Exceptions\JsonException;
 
 use function Safe\json_decode;
@@ -33,17 +35,42 @@ public function isInvalid(): bool
 
     public function connection(): string
     {
-        return $this->task['internal']['connection'];
+        if ($connection = data_get($this->command(), 'connection')) {
+            return $connection;
+        }
+
+        return config('queue.default');
     }
 
     public function queue(): string
     {
-        return $this->task['internal']['queue'];
+        if ($queue = data_get($this->command(), 'queue')) {
+            return $queue;
+        }
+
+        return config('queue.connections.'.$this->connection().'.queue');
     }
 
     public function taskName(): string
     {
-        return $this->task['internal']['taskName'];
+        return request()->header('X-CloudTasks-TaskName')
+            ?? request()->header('X-AppEngine-TaskName')
+            ?? throw new Error('Unable to extract taskname from header');
+    }
+
+    public function command(): array
+    {
+        $command = $this->task['data']['command'];
+
+        if (str_starts_with($command, 'O:')) {
+            return (array) unserialize($command, ['allowed_classes' => false]);
+        }
+
+        if (app()->bound(Encrypter::class)) {
+            return (array) unserialize(app(Encrypter::class)->decrypt($command));
+        }
+
+        return [];
     }
 
     public function toArray(): array

src/Events/TaskIncoming.php

@@ -7,6 +7,7 @@
 use Google\Cloud\Tasks\V2\Client\CloudTasksClient;
 use Illuminate\Container\Container;
 use Illuminate\Queue\WorkerOptions;
+use Stackkit\LaravelGoogleCloudTasksQueue\Events\TaskIncoming;
 
 class TaskHandler
 {
@@ -21,6 +22,8 @@ public function handle(?string $task = null): void
     {
         $task = IncomingTask::fromJson($task ?: request()->getContent());
 
+        event(new TaskIncoming($task));
+
         if ($task->isInvalid()) {
             abort(422, 'Invalid task payload');
         }