fix(prefix): propagate key prefix to Worker, QueueEvents, and FlowProducer

src/server/bymax-queue.module.ts

@@ -83,7 +83,7 @@ export class BymaxQueueModule extends ConfigurableModuleClass {
         {
           provide: FlowService,
           useFactory: (conn: ConnectionResolver): FlowService =>
-            new FlowService(conn, resolved.flows.enabled, resolved.telemetry),
+            new FlowService(conn, resolved.flows.enabled, resolved.prefix, resolved.telemetry),
           inject: [ConnectionResolver],
         },
         {
@@ -128,7 +128,7 @@ export class BymaxQueueModule extends ConfigurableModuleClass {
         {
           provide: FlowService,
           useFactory: (conn: ConnectionResolver, resolved: ResolvedQueueOptions): FlowService =>
-            new FlowService(conn, resolved.flows.enabled, resolved.telemetry),
+            new FlowService(conn, resolved.flows.enabled, resolved.prefix, resolved.telemetry),
           inject: [ConnectionResolver, BYMAX_QUEUE_RESOLVED_OPTIONS],
         },
         {

src/server/services/flow.service.spec.ts

@@ -53,7 +53,7 @@ beforeEach(() => {
 describe('FlowService — enabled', () => {
   it('constructs a FlowProducer on the main connection', () => {
     // When enabled, the producer is built with the resolver's Queue-role client.
-    new FlowService(makeConnection(), true)
+    new FlowService(makeConnection(), true, 'bull')
 
     expect(producerConstructorArgs).toHaveLength(1)
     const [opts] = producerConstructorArgs[0] as [{ connection: unknown }]
@@ -62,7 +62,7 @@ describe('FlowService — enabled', () => {
 
   it('delegates add to the underlying producer', async () => {
     // add forwards the flow definition unchanged and returns the producer result.
-    const service = new FlowService(makeConnection(), true)
+    const service = new FlowService(makeConnection(), true, 'bull')
     const flow = { name: 'root', queueName: 'pdf', data: {} } as FlowJob
     const node = { job: { id: '1' } } as unknown as JobNode
     producerInstances[0]?.add.mockResolvedValue(node)
@@ -73,7 +73,7 @@ describe('FlowService — enabled', () => {
 
   it('delegates addBulk to the underlying producer as a mutable array', async () => {
     // addBulk copies the readonly input into the array BullMQ expects.
-    const service = new FlowService(makeConnection(), true)
+    const service = new FlowService(makeConnection(), true, 'bull')
     const flows = [{ name: 'a', queueName: 'q', data: {} }] as FlowJob[]
     const nodes = [{ job: { id: '1' } }] as unknown as JobNode[]
     producerInstances[0]?.addBulk.mockResolvedValue(nodes)
@@ -84,23 +84,32 @@ describe('FlowService — enabled', () => {
 
   it('returns the underlying producer from getProducer', () => {
     // getProducer exposes the constructed FlowProducer instance.
-    const service = new FlowService(makeConnection(), true)
+    const service = new FlowService(makeConnection(), true, 'bull')
 
     expect(service.getProducer()).toBe(producerInstances[0])
   })
 
   it('passes the configured telemetry into the FlowProducer constructor', () => {
     // A configured telemetry instance reaches the producer so spans propagate to children.
     const telemetry = { name: 'sentinel-telemetry' } as unknown as Telemetry
-    new FlowService(makeConnection(), true, telemetry)
+    new FlowService(makeConnection(), true, 'bull', telemetry)
 
     const [opts] = producerConstructorArgs[0] as [{ telemetry?: unknown }]
     expect(opts.telemetry).toBe(telemetry)
   })
 
+  it('passes the configured prefix into the FlowProducer constructor', () => {
+    // The producer must use the same key prefix as the Queue so flow jobs land
+    // in the keyspace the workers poll; a non-default prefix breaks otherwise.
+    new FlowService(makeConnection(), true, 'tenant:flows')
+
+    const [opts] = producerConstructorArgs[0] as [{ prefix?: unknown }]
+    expect(opts.prefix).toBe('tenant:flows')
+  })
+
   it('omits the telemetry key when telemetry is not configured', () => {
     // Without telemetry, the producer options never carry the key.
-    new FlowService(makeConnection(), true)
+    new FlowService(makeConnection(), true, 'bull')
 
     const [opts] = producerConstructorArgs[0] as [Record<string, unknown>]
     expect('telemetry' in opts).toBe(false)
@@ -110,14 +119,14 @@ describe('FlowService — enabled', () => {
 describe('FlowService — disabled', () => {
   it('does not construct a FlowProducer', () => {
     // When disabled, no producer is built (no Redis connection is opened).
-    new FlowService(makeConnection(), false)
+    new FlowService(makeConnection(), false, 'bull')
 
     expect(producerConstructorArgs).toHaveLength(0)
   })
 
   it('throws FLOW_DISABLED (503) from add', async () => {
     // add is guarded and rejects with the typed disabled error.
-    const service = new FlowService(makeConnection(), false)
+    const service = new FlowService(makeConnection(), false, 'bull')
 
     await expect(service.add({} as FlowJob)).rejects.toBeInstanceOf(QueueException)
     await service.add({} as FlowJob).catch((err: unknown) => {
@@ -128,14 +137,14 @@ describe('FlowService — disabled', () => {
 
   it('throws FLOW_DISABLED from addBulk', async () => {
     // addBulk is guarded the same way as add.
-    const service = new FlowService(makeConnection(), false)
+    const service = new FlowService(makeConnection(), false, 'bull')
 
     await expect(service.addBulk([])).rejects.toBeInstanceOf(QueueException)
   })
 
   it('throws FLOW_DISABLED from getProducer', () => {
     // getProducer cannot return a producer that was never created.
-    const service = new FlowService(makeConnection(), false)
+    const service = new FlowService(makeConnection(), false, 'bull')
 
     expect(() => service.getProducer()).toThrow(QueueException)
   })
@@ -144,7 +153,7 @@ describe('FlowService — disabled', () => {
 describe('FlowService — onModuleDestroy', () => {
   it('closes the producer when active', async () => {
     // Shutdown closes the active producer a single time.
-    const service = new FlowService(makeConnection(), true)
+    const service = new FlowService(makeConnection(), true, 'bull')
 
     await service.onModuleDestroy()
 
@@ -153,15 +162,15 @@ describe('FlowService — onModuleDestroy', () => {
 
   it('swallows a close rejection', async () => {
     // A failing close must not abort the shutdown sequence.
-    const service = new FlowService(makeConnection(), true)
+    const service = new FlowService(makeConnection(), true, 'bull')
     producerInstances[0]?.close.mockRejectedValue(new Error('close failed'))
 
     await expect(service.onModuleDestroy()).resolves.toBeUndefined()
   })
 
   it('is a no-op when inactive', async () => {
     // With no producer there is nothing to close.
-    const service = new FlowService(makeConnection(), false)
+    const service = new FlowService(makeConnection(), false, 'bull')
 
     await expect(service.onModuleDestroy()).resolves.toBeUndefined()
     expect(producerInstances).toHaveLength(0)

src/server/services/flow.service.ts

@@ -33,10 +33,18 @@ import { QUEUE_ERROR_CODES } from '../constants/error-codes'
 export class FlowService implements OnModuleDestroy {
   private readonly producer?: FlowProducer
 
-  constructor(connection: ConnectionResolver, enabled: boolean, telemetry?: Telemetry) {
+  constructor(
+    connection: ConnectionResolver,
+    enabled: boolean,
+    prefix: string,
+    telemetry?: Telemetry,
+  ) {
     if (enabled) {
       this.producer = new FlowProducer({
         connection: connection.getClient(),
+        // Match the producer Queue's key prefix so flow jobs land in the same
+        // keyspace the workers poll; a non-default prefix breaks flows otherwise.
+        prefix,
         ...(telemetry ? { telemetry } : {}),
       })
     }

src/server/services/queue-events-registry.service.spec.ts

@@ -6,6 +6,7 @@
 
 import { QueueEventsRegistry } from './queue-events-registry.service'
 import type { ConnectionResolver } from './connection-resolver.service'
+import type { ResolvedQueueOptions } from '../config/resolved-options'
 import type { Redis } from 'ioredis'
 
 /** Track instances created by the mocked constructor. */
@@ -37,14 +38,19 @@ function fakeConnection(redis?: Redis): ConnectionResolver {
   } as unknown as ConnectionResolver
 }
 
+/** Minimal resolved-options stub carrying just the key prefix. */
+function fakeOptions(prefix = 'bull'): ResolvedQueueOptions {
+  return { prefix } as unknown as ResolvedQueueOptions
+}
+
 beforeEach(() => {
   createdQueueEvents.length = 0
 })
 
 describe('QueueEventsRegistry.getOrCreate', () => {
   it('creates a QueueEvents on first call and returns it', () => {
     // First call must instantiate and cache a QueueEvents for the queue.
-    const registry = new QueueEventsRegistry(fakeConnection())
+    const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions())
     const qe = registry.getOrCreate('email')
     expect(qe).toBeDefined()
     expect(createdQueueEvents).toHaveLength(1)
@@ -55,10 +61,22 @@ describe('QueueEventsRegistry.getOrCreate', () => {
     // SUBSCRIBE needs maxRetriesPerRequest: null), not an empty/default options bag.
     const { QueueEvents: MockQueueEvents } = jest.requireMock<{ QueueEvents: jest.Mock }>('bullmq')
     const redis = fakeRedis()
-    const registry = new QueueEventsRegistry(fakeConnection(redis))
+    const registry = new QueueEventsRegistry(fakeConnection(redis), fakeOptions())
     registry.getOrCreate('email')
     const dup = (redis.duplicate as jest.Mock).mock.results[0]!.value as unknown
-    expect(MockQueueEvents).toHaveBeenCalledWith('email', { connection: dup })
+    expect(MockQueueEvents).toHaveBeenCalledWith('email', { connection: dup, prefix: 'bull' })
+  })
+
+  it('constructs QueueEvents with the configured key prefix', () => {
+    // QueueEvents must watch the same prefixed keyspace the producer writes to;
+    // a non-default prefix would otherwise silently drop every event.
+    const { QueueEvents: MockQueueEvents } = jest.requireMock<{ QueueEvents: jest.Mock }>('bullmq')
+    const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions('tenant:events'))
+    registry.getOrCreate('email')
+    expect(MockQueueEvents).toHaveBeenCalledWith(
+      'email',
+      expect.objectContaining({ prefix: 'tenant:events' }),
+    )
   })
 
   it('disconnects the duplicated connection when QueueEvents constructor throws (no leak)', () => {
@@ -70,7 +88,7 @@ describe('QueueEventsRegistry.getOrCreate', () => {
       throw new Error('subscribe failed')
     })
     const redis = fakeRedis()
-    const registry = new QueueEventsRegistry(fakeConnection(redis))
+    const registry = new QueueEventsRegistry(fakeConnection(redis), fakeOptions())
     expect(() => registry.getOrCreate('email')).toThrow('subscribe failed')
     const dupResult = (redis.duplicate as jest.Mock).mock.results[0]!.value as {
       disconnect: jest.Mock
@@ -80,7 +98,7 @@ describe('QueueEventsRegistry.getOrCreate', () => {
 
   it('returns the same instance on subsequent calls (idempotent)', () => {
     // Idempotency is critical — no second Redis connection must be opened.
-    const registry = new QueueEventsRegistry(fakeConnection())
+    const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions())
     const first = registry.getOrCreate('email')
     const second = registry.getOrCreate('email')
     expect(first).toBe(second)
@@ -89,7 +107,7 @@ describe('QueueEventsRegistry.getOrCreate', () => {
 
   it('creates separate instances for different queues', () => {
     // Each queue requires its own QueueEvents (and its own Redis connection).
-    const registry = new QueueEventsRegistry(fakeConnection())
+    const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions())
     const emailQe = registry.getOrCreate('email')
     const smsQe = registry.getOrCreate('sms')
     expect(emailQe).not.toBe(smsQe)
@@ -100,15 +118,15 @@ describe('QueueEventsRegistry.getOrCreate', () => {
 describe('QueueEventsRegistry.list / getAll', () => {
   it('list returns the names of queues with an open QueueEvents', () => {
     // list() surfaces which queues currently have a QueueEvents connection.
-    const registry = new QueueEventsRegistry(fakeConnection())
+    const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions())
     registry.getOrCreate('email')
     registry.getOrCreate('sms')
     expect(registry.list()).toEqual(expect.arrayContaining(['email', 'sms']))
   })
 
   it('getAll returns a ReadonlyMap of all QueueEvents', () => {
     // getAll() is used by the shutdown orchestrator.
-    const registry = new QueueEventsRegistry(fakeConnection())
+    const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions())
     registry.getOrCreate('email')
     const map = registry.getAll()
     expect(map.size).toBe(1)
@@ -119,7 +137,7 @@ describe('QueueEventsRegistry.list / getAll', () => {
 describe('QueueEventsRegistry.getConnections', () => {
   it('exposes the duplicated connection opened per queue', () => {
     // The shutdown orchestrator closes exactly these library-created connections.
-    const registry = new QueueEventsRegistry(fakeConnection())
+    const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions())
     registry.getOrCreate('email')
     registry.getOrCreate('sms')
     const connections = registry.getConnections()
@@ -130,7 +148,7 @@ describe('QueueEventsRegistry.getConnections', () => {
 
   it('returns an empty map before any QueueEvents are created', () => {
     // With nothing observed there are no library-created connections to close.
-    const registry = new QueueEventsRegistry(fakeConnection())
+    const registry = new QueueEventsRegistry(fakeConnection(), fakeOptions())
     expect(registry.getConnections().size).toBe(0)
   })
 })

src/server/services/queue-events-registry.service.ts

@@ -11,6 +11,8 @@ import { Inject, Injectable } from '@nestjs/common'
 import { QueueEvents } from 'bullmq'
 import type { Redis } from 'ioredis'
 import { ConnectionResolver } from './connection-resolver.service'
+import { BYMAX_QUEUE_RESOLVED_OPTIONS } from '../bymax-queue.constants'
+import type { ResolvedQueueOptions } from '../config/resolved-options'
 import { duplicateConnection } from '../utils/duplicate-connection'
 
 /**
@@ -34,7 +36,10 @@ export class QueueEventsRegistry {
   private readonly events = new Map<string, QueueEvents>()
   private readonly connections = new Map<string, Redis>()
 
-  constructor(@Inject(ConnectionResolver) private readonly connection: ConnectionResolver) {}
+  constructor(
+    @Inject(ConnectionResolver) private readonly connection: ConnectionResolver,
+    @Inject(BYMAX_QUEUE_RESOLVED_OPTIONS) private readonly options: ResolvedQueueOptions,
+  ) {}
 
   /**
    * Returns the cached `QueueEvents` for `queueName`, creating one on first
@@ -50,7 +55,7 @@ export class QueueEventsRegistry {
     const conn = duplicateConnection(this.connection.getClient())
     let qe: QueueEvents
     try {
-      qe = new QueueEvents(queueName, { connection: conn })
+      qe = new QueueEvents(queueName, { connection: conn, prefix: this.options.prefix })
     } catch (err) {
       // Disconnect the duplicated connection to prevent a Redis resource leak.
       conn.disconnect()

src/server/services/worker-registry.service.spec.ts

@@ -91,6 +91,15 @@ describe('WorkerRegistry.register', () => {
     expect(registry.list()).toContain('email')
   })
 
+  it('constructs the Worker with the configured key prefix', () => {
+    // The worker must poll the same prefixed keyspace the producer Queue writes
+    // to; a non-default prefix would otherwise leave every job unconsumed.
+    const registry = new WorkerRegistry(fakeConnection(), { ...makeOptions(), prefix: 'tenant:wk' })
+    registry.register({ queueName: 'email', handler: noopHandler })
+    const [, , workerOpts] = workerConstructorArgs[0] as [unknown, unknown, { prefix?: unknown }]
+    expect(workerOpts.prefix).toBe('tenant:wk')
+  })
+
   it('throws DUPLICATE_PROCESSOR when the same queue is registered twice', () => {
     // Two processors targeting the same queue is an application-level error.
     const registry = new WorkerRegistry(fakeConnection(), makeOptions())

src/server/services/worker-registry.service.ts

@@ -360,6 +360,9 @@ export class WorkerRegistry {
       connection: conn,
       concurrency: opts?.concurrency ?? DEFAULT_WORKER_CONCURRENCY,
       autorun: opts?.autorun ?? true,
+      // Match the producer Queue's key prefix so the worker polls the same
+      // keyspace; without this a non-default prefix leaves jobs unconsumed.
+      prefix: this.options.prefix,
     }
     if (opts?.limiter !== undefined) result.limiter = opts.limiter
     if (opts?.lockDuration !== undefined) result.lockDuration = opts.lockDuration