Add metadata envelope to queue messages

composer.json

@@ -29,6 +29,7 @@
     "require-dev": {
         "cakephp/bake": "^3.5.1",
         "cakephp/cakephp-codesniffer": "^5.3",
+        "enqueue/enqueue": "^0.10.27",
         "enqueue/fs": "^0.10",
         "phpunit/phpunit": "^11.5.3 || ^12.1.3 || ^13.0"
     },

config/Migrations/20260922163025_AddMetadataToFailedJobs.php

@@ -0,0 +1,33 @@
+<?php
+declare(strict_types=1);
+
+use Migrations\BaseMigration;
+
+class AddMetadataToFailedJobs extends BaseMigration
+{
+    /**
+     * Add nullable envelope metadata so failed jobs keep bookkeeping fields
+     * (`tags`, `_uniqueId`, `batch_id`) through store and requeue. Nullable
+     * for backward compatibility with existing rows and legacy messages
+     * dispatched without a metadata envelope.
+     *
+     * Also converts `data` to a native `json` column so the ORM encodes/decodes
+     * it automatically instead of relying on the manual JSON handling of the
+     * legacy `text` column.
+     *
+     * @return void
+     */
+    public function change(): void
+    {
+        $table = $this->table('queue_failed_jobs');
+        $table->addColumn('metadata', 'json', [
+                'null' => true,
+                'default' => null,
+            ])
+            ->changeColumn('data', 'json', [
+                'null' => false,
+                'default' => null,
+            ])
+            ->update();
+    }
+}

docs/en/failed-jobs.md

@@ -11,6 +11,16 @@ If `storeFailedJobs` is enabled, failed jobs are rejected, written to the `queue
 
 Your broker may also support dead-letter queues. Those are separate transport-level features. Queue's failed-job storage is transport-agnostic and only tracks jobs that exhaust Queue's retry handling.
 
+## Metadata Envelope
+
+The `queue_failed_jobs` table has a nullable `metadata` column that preserves the message envelope (`tags`, `_uniqueId`, `batch_id`, ...) through store and requeue: `queue requeue` pushes `data` and `metadata` back together. Rows stored before the envelope existed have `NULL` metadata and requeue exactly as before. Make sure the plugin migrations are applied:
+
+```bash
+bin/cake migrations migrate --plugin Cake/Queue
+```
+
+Both `data` and `metadata` are native `json` columns, so the ORM encodes and decodes them automatically — `$failedJob->data` and `$failedJob->metadata` are already arrays, no manual `json_decode()` needed.
+
 ## Requeue Failed Jobs
 
 Push failed jobs back onto the queue:

docs/en/jobs.md

@@ -42,6 +42,7 @@ Job classes can use constructor injection in the same way as controllers or comm
 The `Message` object exposes:
 
 - `getArgument($key = null, $default = null)` to fetch the full payload or a nested value using `Hash::get()` notation.
+- `getMetadata()` to read the envelope metadata recorded next to the payload (see "Message Metadata Envelope").
 - `getContext()` to access the original queue context.
 - `getOriginalMessage()` to access the broker-specific message object.
 - `getParsedBody()` to inspect the parsed queue body.
@@ -87,6 +88,7 @@ Supported options:
 - `config`: queue config name. Defaults to `default`.
 - `delay`: delay processing by a number of seconds. Broker support varies.
 - `expires`: expire the message after a number of seconds if it has not been consumed.
+- `metadata`: optional envelope array recorded on the message body next to `data` (e.g. `tags`, `_uniqueId`, `batch_id`). Unlike `data` it is never passed to the job or hydrated into a DTO. Omitted from the body when empty. Defaults to `[]`.
 - `priority`: one of `\Enqueue\Client\MessagePriority::VERY_LOW`, `LOW`, `NORMAL`, `HIGH`, or `VERY_HIGH`.
 - `queue`: queue name to use. Defaults to the configured queue, then `default`.
 
@@ -155,3 +157,34 @@ Hydration mirrors the DTO conventions used elsewhere in CakePHP (`#[RequestToDto
   ```
 
 - **`createFromArray()` factory** — if the DTO class defines a static `createFromArray(array $data, bool $nested = false): static` method, it's used instead of reflection.
+
+## Message Metadata Envelope
+
+Besides the payload (`data`), a message can carry an envelope of bookkeeping fields in a sibling `metadata` section:
+
+```php
+QueueManager::push(ProcessOrderJob::class, $order, [
+    'metadata' => ['tags' => ['finance'], '_uniqueId' => $uuid],
+]);
+```
+
+```php
+public function execute(Message $message): ?string
+{
+    $order = $message->getDto(OrderDto::class);
+    $tags = $message->getMetadata()['tags'] ?? [];
+
+    return Processor::ACK;
+}
+```
+
+Rules:
+
+- `metadata` travels next to `data`, never inside it, so the payload stays pure and DTO-safe.
+- It is never passed to the job and never hydrated into a DTO — read it with `Message::getMetadata()`, which returns `[]` for legacy messages without an envelope.
+- It is omitted from the message body when empty, so pushes without it produce byte-identical bodies.
+- It survives failed-job storage and requeue (see Failed Jobs).
+
+### Third-party payload
+
+The envelope is intentionally opaque to this plugin: it is a carrier for third-party producer and monitoring software, which may write bookkeeping fields at dispatch time and read them back on the worker side. The base plugin transports and persists the envelope but never interprets its keys — what a tag means, and which keys matter, is defined by those consumers.

docs/en/upgrading.md

@@ -10,3 +10,11 @@ Queue 3.0 requires PHP 8.2 or later and CakePHP 5.4 or later.
 ```php
 $this->addPlugin('Cake/Queue');
 ```
+
+## Failed-Job `data` Is Now a Native JSON Column
+
+The `queue_failed_jobs.data` column (and the new `metadata` column, see Failed Jobs) uses the native `json` type, so the ORM returns arrays directly: `$failedJob->data` is an array, not a JSON string. The `FailedJob::decoded_data` and `FailedJob::decoded_metadata` accessors were removed — read `$failedJob->data` and `$failedJob->metadata` instead. Run the plugin migrations to get the new schema:
+
+```bash
+bin/cake migrations migrate --plugin Cake/Queue
+```

src/Command/RequeueCommand.php

@@ -133,12 +133,13 @@ public function execute(Arguments $args, ConsoleIo $io): int
             try {
                 QueueManager::push(
                     [$failedJob->class, $failedJob->method],
-                    $failedJob->decoded_data,
+                    $failedJob->data,
                     [
                         'config' => $failedJob->config,
                         'priority' => $failedJob->priority,
                         'queue' => $failedJob->queue,
                         'dtoClass' => $failedJob->dto_class ?? null,
+                        'metadata' => $failedJob->metadata ?? [],
                     ],
                 );
 

src/Job/Message.php

@@ -204,6 +204,21 @@ public function getDto(string $dtoClass): object
         return $dto;
     }
 
+    /**
+     * Get the envelope metadata recorded on the message body at dispatch time.
+     *
+     * Metadata carries bookkeeping fields that belong next to the payload,
+     * not inside it, so `data` stays pure for DTO hydration.
+     *
+     * @return array<string, mixed>
+     */
+    public function getMetadata(): array
+    {
+        $metadata = $this->parsedBody['metadata'] ?? [];
+
+        return is_array($metadata) ? $metadata : [];
+    }
+
     /**
      * The maximum number of attempts allowed by the job.
      */

src/Listener/FailedJobsListener.php

@@ -66,7 +66,8 @@ public function storeFailedJob(object $event): void
         $failedJob = $failedJobsTable->newEntity([
             'class' => $class,
             'method' => $method,
-            'data' => json_encode($data),
+            'data' => $data,
+            'metadata' => $originalMessageBody['metadata'] ?? null,
             'dto_class' => $originalMessageBody['dtoClass'] ?? null,
             'config' => $requeueOptions['config'],
             'priority' => $requeueOptions['priority'],

src/Model/Entity/FailedJob.php

@@ -11,15 +11,14 @@
  * @property int $id
  * @property string $class
  * @property string $method
- * @property string $data
+ * @property array<string, mixed> $data
+ * @property array<string, mixed>|null $metadata
  * @property string|null $dto_class
  * @property string|null $config
  * @property string|null $priority
  * @property string|null $queue
  * @property string|null $exception
  * @property \Cake\I18n\DateTime|null $created
- *
- * @property array $decoded_data
  */
 class FailedJob extends Entity
 {
@@ -39,19 +38,11 @@ class FailedJob extends Entity
         'method' => true,
         'data' => true,
         'dto_class' => true,
+        'metadata' => true,
         'config' => true,
         'priority' => true,
         'queue' => true,
         'exception' => true,
         'created' => true,
     ];
-
-    /**
-     * @see \Cake\Queue\Model\Entity\FailedJob::$decoded_data
-     * @return array<string, mixed>
-     */
-    protected function _getDecodedData(): array
-    {
-        return json_decode($this->data, true);
-    }
 }

src/Model/Table/FailedJobsTable.php

@@ -40,6 +40,14 @@ public function initialize(array $config): void
         $this->setDisplayField('id');
         $this->setPrimaryKey('id');
 
+        // Force the JSON type on every driver. SQLite stores json columns as
+        // text, so introspection alone would not apply Cake\ORM's JsonType.
+        $schema = $this->getSchema();
+        $schema->setColumnType('data', 'json');
+        if ($schema->hasColumn('metadata')) {
+            $schema->setColumnType('metadata', 'json');
+        }
+
         $this->addBehavior('Timestamp');
     }
 
@@ -68,9 +76,13 @@ public function validationDefault(Validator $validator): Validator
             ->notEmptyString('method');
 
         $validator
-            ->scalar('data')
             ->requirePresence('data', 'create')
-            ->notEmptyString('data');
+            ->allowEmptyArray('data')
+            ->array('data');
+
+        $validator
+            ->allowEmptyArray('metadata')
+            ->array('metadata');
 
         $validator
             ->scalar('dto_class')

src/QueueManager.php

@@ -223,6 +223,10 @@ public static function engine(string $name): SimpleClient
      *   - `expires` - Time (in integer seconds) after which the message expires.
      *     The message will be removed from the queue if this time is exceeded
      *     and it has not been consumed. Default `null`.
+     *   - `metadata` - Optional envelope data recorded on the message body next
+     *     to `data` (e.g. `tags`, `_uniqueId`, `batch_id`). Unlike `data` it is
+     *     never passed to the job or hydrated into a DTO. Omitted from the body
+     *     when empty or not an array. Default `[]`.
      *   - `priority` - Valid values:
      *      - `\Enqueue\Client\MessagePriority::VERY_LOW`
      *      - `\Enqueue\Client\MessagePriority::LOW`
@@ -297,6 +301,11 @@ public static function push(string|array $className, array|object $data = [], ar
             $body['dtoClass'] = $dtoClass;
         }
 
+        $metadata = $options['metadata'] ?? null;
+        if (is_array($metadata) && $metadata !== []) {
+            $body['metadata'] = $metadata;
+        }
+
         $message = new ClientMessage($body);
 
         if (isset($options['delay'])) {

src/TestSuite/QueueTrait.php

@@ -241,6 +241,66 @@ public function assertJobQueuedWithPriority(
         );
     }
 
+    /**
+     * Assert a job was queued with specific metadata
+     *
+     * @param string $jobClass Job class name
+     * @param array<string, mixed> $metadata Expected metadata
+     * @param string $message Optional assertion message
+     * @return void
+     */
+    public function assertJobQueuedWithMetadata(
+        string $jobClass,
+        array $metadata,
+        string $message = '',
+    ): void {
+        $jobs = TestQueueClient::getQueuedJobsByClass($jobClass);
+        $this->assertNotEmpty($jobs, sprintf('Job %s was not queued', $jobClass));
+
+        $found = false;
+        foreach ($jobs as $job) {
+            if (($job['metadata'] ?? []) === $metadata) {
+                $found = true;
+                break;
+            }
+        }
+
+        $this->assertTrue(
+            $found,
+            $message ?: sprintf('Job %s was not queued with expected metadata', $jobClass),
+        );
+    }
+
+    /**
+     * Assert a job was queued with a specific DTO class
+     *
+     * @param string $jobClass Job class name
+     * @param string $dtoClass Expected DTO class name
+     * @param string $message Optional assertion message
+     * @return void
+     */
+    public function assertJobQueuedWithDtoClass(
+        string $jobClass,
+        string $dtoClass,
+        string $message = '',
+    ): void {
+        $jobs = TestQueueClient::getQueuedJobsByClass($jobClass);
+        $this->assertNotEmpty($jobs, sprintf('Job %s was not queued', $jobClass));
+
+        $found = false;
+        foreach ($jobs as $job) {
+            if (($job['dtoClass'] ?? null) === $dtoClass) {
+                $found = true;
+                break;
+            }
+        }
+
+        $this->assertTrue(
+            $found,
+            $message ?: sprintf('Job %s was not queued with DTO class %s', $jobClass, $dtoClass),
+        );
+    }
+
     /**
      * Assert a job was queued a specific number of times
      *

src/TestSuite/TestQueueClient.php

@@ -108,6 +108,15 @@ public static function captureMessage(
         $jobClass = is_array($classData) ? $classData[0] : null;
         $method = is_array($classData) && isset($classData[1]) ? $classData[1] : 'execute';
         $data = $body['data'] ?? ($body['args'][0] ?? []);
+        $dtoClass = $body['dtoClass'] ?? null;
+        if (!is_string($dtoClass)) {
+            $dtoClass = null;
+        }
+
+        $metadata = $body['metadata'] ?? [];
+        if (!is_array($metadata)) {
+            $metadata = [];
+        }
 
         $requeueOptions = $body['requeueOptions'] ?? [];
         $configName = $requeueOptions['config'] ?? 'default';
@@ -145,6 +154,8 @@ public static function captureMessage(
             'jobClass' => $jobClass,
             'method' => $method,
             'data' => $data,
+            'dtoClass' => $dtoClass,
+            'metadata' => $metadata,
             'options' => [
                 'config' => $configName,
                 'queue' => $queueName,

tests/Fixture/FailedJobsFixture.php

@@ -26,7 +26,8 @@ public function init(): void
                 'id' => 1,
                 'class' => LogToDebugJob::class,
                 'method' => 'execute',
-                'data' => '{"sample_data_1": "sample value", "sample_data_2": 1}',
+                'data' => ['sample_data_1' => 'sample value', 'sample_data_2' => 1],
+                'metadata' => null,
                 'dto_class' => null,
                 'config' => 'default',
                 'priority' => null,
@@ -38,7 +39,8 @@ public function init(): void
                 'id' => 2,
                 'class' => MaxAttemptsIsThreeJob::class,
                 'method' => 'execute',
-                'data' => '{"sample_data_1": "sample value", "sample_data_2": 1}',
+                'data' => ['sample_data_1' => 'sample value', 'sample_data_2' => 1],
+                'metadata' => null,
                 'dto_class' => null,
                 'config' => 'default',
                 'priority' => null,
@@ -50,7 +52,8 @@ public function init(): void
                 'id' => 3,
                 'class' => LogToDebugJob::class,
                 'method' => 'execute',
-                'data' => '{"sample_data_1": "sample value", "sample_data_2": 1}',
+                'data' => ['sample_data_1' => 'sample value', 'sample_data_2' => 1],
+                'metadata' => null,
                 'dto_class' => null,
                 'config' => 'alternate_config',
                 'priority' => null,
@@ -60,5 +63,8 @@ public function init(): void
             ],
         ];
         parent::init();
+
+        $this->_schema->setColumnType('data', 'json');
+        $this->_schema->setColumnType('metadata', 'json');
     }
 }