feat: add support to auto inject context data into workers

configurer/worker

@@ -1,20 +1,4 @@
 export default {
-  /*
-  |--------------------------------------------------------------------------
-  | Configurations for cls-rtracer plugin.
-  |--------------------------------------------------------------------------
-  |
-  | This values defines all the configurations for cls-rtracer plugins. Check
-  | the documentation for more information:
-  |
-  | https://github.com/puzpuzpuz/cls-rtracer
-  |
-  */
-
-  rTracer: {
-    enabled: true
-  },
-
   /*
   |--------------------------------------------------------------------------
   | Log worker tasks 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@athenna/queue",
-  "version": "5.30.0",
+  "version": "5.31.0",
   "description": "The Athenna queue handler.",
   "license": "MIT",
   "author": "João Lenon <lenon@athenna.io>",

src/drivers/AwsSqsDriver.ts

@@ -23,6 +23,7 @@ import { Is, Options, Uuid } from '@athenna/common'
 import type { ConnectionOptions } from '#src/types'
 import { ConnectionFactory } from '#src/factories/ConnectionFactory'
 import { QueueExecutionScope } from '#src/worker/QueueExecutionScope'
+import { QueueJobPropagationHelper } from '#src/helpers/QueueJobPropagationHelper'
 import { AwsSqsDriverExceptionHandler } from '#src/handlers/AwsSqsDriverExceptionHandler'
 import { NotFifoSqsQueueTypeException } from '#src/exceptions/NotFifoSqsQueueTypeException'
 
@@ -433,14 +434,16 @@ export class AwsSqsDriver extends Driver<SQSClient> {
       data: job.data
     }
 
+    const executionJob = QueueJobPropagationHelper.getJob(workerJob)
+
     await this.runScopedQueueProcessor(
       processor,
       workerJob,
       async () => {
         try {
           startHeartbeat()
 
-          await processor(workerJob)
+          await processor(executionJob)
 
           stopHeartbeat()
 

src/drivers/DatabaseDriver.ts

@@ -13,6 +13,7 @@ import { Is, Options } from '@athenna/common'
 import type { ConnectionOptions } from '#src/types'
 import type { DatabaseImpl } from '@athenna/database'
 import { ConnectionFactory } from '#src/factories/ConnectionFactory'
+import { QueueJobPropagationHelper } from '#src/helpers/QueueJobPropagationHelper'
 import { DatabaseDriverExceptionHandler } from '#src/handlers/DatabaseDriverExceptionHandler'
 
 export class DatabaseDriver extends Driver<DatabaseImpl> {
@@ -331,10 +332,11 @@ export class DatabaseDriver extends Driver<DatabaseImpl> {
       attempts: job.attempts,
       data: job.data
     }
+    const executionJob = QueueJobPropagationHelper.getJob(workerJob)
 
     await this.runScopedQueueProcessor(processor, workerJob, async () => {
       try {
-        await processor(workerJob)
+        await processor(executionJob)
 
         /**
          * If the job still exists after processing, it means that the job was

src/drivers/Driver.ts

@@ -12,6 +12,7 @@ import { Is } from '@athenna/common'
 import { Config } from '@athenna/config'
 import type { Job, ConnectionOptions } from '#src/types'
 import { QueueExecutionScope } from '#src/worker/QueueExecutionScope'
+import { QueueJobPropagationHelper } from '#src/helpers/QueueJobPropagationHelper'
 
 export const RUN_WITH_WORKER_CONTEXT = Symbol.for(
   '@athenna/queue.runWithWorkerContext'
@@ -199,13 +200,19 @@ export abstract class Driver<Client = any> {
       return runner(data, callback, captureScope)
     }
 
-    const scope = new QueueExecutionScope<T>({
-      name: this.queueName,
-      connection: this.connection,
-      options: this.options,
-      traceId: null,
-      job: this.createContextJob(data)
-    })
+    const scope = new QueueExecutionScope<T>(
+      {
+        name: this.queueName,
+        connection: this.connection,
+        options: this.options,
+        traceId: null,
+        job: QueueJobPropagationHelper.getJob(this.createContextJob(data))
+      },
+      {
+        carrier: this.getJobCarrier(data),
+        currentContextValues: this.getJobCurrentContextValues(data)
+      }
+    )
 
     captureScope?.(scope)
 
@@ -224,6 +231,22 @@ export abstract class Driver<Client = any> {
     } as Job
   }
 
+  private getJobCarrier<T>(data: T) {
+    if (!this.isJob(data)) {
+      return {}
+    }
+
+    return QueueJobPropagationHelper.getCarrier(data.data)
+  }
+
+  private getJobCurrentContextValues<T>(data: T) {
+    if (!this.isJob(data)) {
+      return {}
+    }
+
+    return QueueJobPropagationHelper.getCurrentContextValues(data.data)
+  }
+
   private isJob(data: unknown): data is Job {
     if (!data || !Is.Object(data)) {
       return false

src/drivers/FakeDriver.ts

@@ -12,6 +12,7 @@ import { Is, Json, Options } from '@athenna/common'
 import type { Job, ConnectionOptions } from '#src/types'
 import { ConnectionFactory } from '#src/factories/ConnectionFactory'
 import { QueueExecutionScope } from '#src/worker/QueueExecutionScope'
+import { QueueJobPropagationHelper } from '#src/helpers/QueueJobPropagationHelper'
 import {
   RUN_WITH_WORKER_CONTEXT,
   type ScopedQueueProcessor
@@ -171,8 +172,8 @@ export class FakeDriver {
    * const user = await Queue.pop()
    * ```
    */
-  public static async pop(): Promise<any> {
-    return {}
+  public static async pop<T = any>(): Promise<T> {
+    return null as T
   }
 
   /**
@@ -185,8 +186,8 @@ export class FakeDriver {
    * const user = await Queue.pop()
    * ```
    */
-  public static async peek(): Promise<any> {
-    return {}
+  public static async peek<T = any>(): Promise<T> {
+    return null as T
   }
 
   /**
@@ -282,13 +283,19 @@ export class FakeDriver {
       return runner(data, callback, captureScope)
     }
 
-    const scope = new QueueExecutionScope<T>({
-      name: this.queueName,
-      connection: this.connection,
-      options: this.options,
-      traceId: null,
-      job: this.createContextJob(data)
-    })
+    const scope = new QueueExecutionScope<T>(
+      {
+        name: this.queueName,
+        connection: this.connection,
+        options: this.options,
+        traceId: null,
+        job: QueueJobPropagationHelper.getJob(this.createContextJob(data))
+      },
+      {
+        carrier: this.getJobCarrier(data),
+        currentContextValues: this.getJobCurrentContextValues(data)
+      }
+    )
 
     captureScope?.(scope)
 
@@ -311,10 +318,13 @@ export class FakeDriver {
     processor: (data: unknown) => any | Promise<any>
   ) {
     const data = await this.pop()
+    const executionData = this.isJob(data)
+      ? QueueJobPropagationHelper.getJob(data)
+      : data
 
     await this.runScopedQueueProcessor(processor, data, async () => {
       try {
-        await processor(data)
+        await processor(executionData)
       } catch (err) {
         Log.channelOrVanilla('exception').error({
           msg: `failed to process job: ${err.message}`,
@@ -331,4 +341,20 @@ export class FakeDriver {
       }
     })
   }
+
+  private static getJobCarrier<T>(data: T) {
+    if (!this.isJob(data)) {
+      return {}
+    }
+
+    return QueueJobPropagationHelper.getCarrier(data.data)
+  }
+
+  private static getJobCurrentContextValues<T>(data: T) {
+    if (!this.isJob(data)) {
+      return {}
+    }
+
+    return QueueJobPropagationHelper.getCurrentContextValues(data.data)
+  }
 }

src/drivers/MemoryDriver.ts

@@ -11,6 +11,7 @@ import { Driver } from '#src/drivers/Driver'
 import { Options, Uuid } from '@athenna/common'
 import type { ConnectionOptions } from '#src/types'
 import { ConnectionFactory } from '#src/factories/ConnectionFactory'
+import { QueueJobPropagationHelper } from '#src/helpers/QueueJobPropagationHelper'
 import { MemoryDriverExceptionHandler } from '#src/handlers/MemoryDriverExceptionHandler'
 
 export class MemoryDriver extends Driver {
@@ -269,10 +270,11 @@ export class MemoryDriver extends Driver {
       attempts: job.attempts,
       data: job.data
     }
+    const executionJob = QueueJobPropagationHelper.getJob(workerJob)
 
     await this.runScopedQueueProcessor(processor, workerJob, async () => {
       try {
-        await processor(workerJob)
+        await processor(executionJob)
 
         /**
          * If the job still exists after processing, it means that the job was