feat: support @athenna/otel

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@athenna/event",
-  "version": "5.5.0",
+  "version": "5.7.0",
   "description": "The Athenna events handler with queue store. Based on Emittery syntax.",
   "license": "MIT",
   "author": "João Lenon <lenon@athenna.io>",

src/events/EventImpl.ts

@@ -8,10 +8,12 @@
  */
 
 import { Config } from '@athenna/config'
-import { Is, Macroable } from '@athenna/common'
 import { Listener } from '#src/events/Listener'
-import { QueueImpl, type ConnectionOptions } from '@athenna/queue'
+import { Is, Macroable, Module } from '@athenna/common'
 import type { EventClosure, Context } from '#src/types'
+import { QueueImpl, type ConnectionOptions } from '@athenna/queue'
+
+const otelModule = await Module.safeImport('@athenna/otel')
 
 export class EventImpl extends Macroable {
   /**
@@ -460,12 +462,23 @@ export class EventImpl extends Macroable {
         emittedAt: data.emittedAt
       }
 
-      await record.closure(ctx)
+      await this.runWithOtelContext(ctx, () => record.closure(ctx))
 
       await queue.ack(job.id)
     })
   }
 
+  private runWithOtelContext<T>(ctx: Context, callback: () => T): T {
+    if (!Config.is('event.otel.contextEnabled', true) || !otelModule) {
+      return callback()
+    }
+
+    return otelModule.Otel.withContext(callback, {
+      bindings: Config.get('event.otel.contextBindings', []),
+      resolveBinding: binding => binding.resolve(ctx)
+    })
+  }
+
   private parseConnectionName(con: string) {
     if (con === 'default') {
       return Config.get('event.store')

tests/unit/events/EventImplTest.ts

@@ -291,6 +291,36 @@ export class EventImplTest {
     assert.equal(specific, 1)
   }
 
+  @Test()
+  public async shouldRunQueuedListenersThroughOtelContextWrapper({ assert }: Context) {
+    const event = Event.store('memoryA') as any
+    const originalRunWithOtelContext = event.runWithOtelContext?.bind(event)
+    let wrappedCtx: any = null
+    let closureCtx: any = null
+
+    event.runWithOtelContext = (ctx, callback) => {
+      wrappedCtx = ctx
+
+      return callback()
+    }
+
+    try {
+      event.on('otel:event', ctx => {
+        closureCtx = ctx
+      })
+
+      await event.emit('otel:event', { ok: true })
+      await Sleep.for(40).milliseconds().wait()
+    } finally {
+      event.runWithOtelContext = originalRunWithOtelContext
+    }
+
+    assert.isDefined(wrappedCtx)
+    assert.deepEqual(wrappedCtx, closureCtx)
+    assert.equal(wrappedCtx.event, 'otel:event')
+    assert.deepEqual(wrappedCtx.data, { ok: true })
+  }
+
   @Test()
   public async shouldBeAbleToEmitSequentially({ assert }: Context) {
     const event = Event.store('memoryA')