refactor!: extract queue config resolution from QueueManager

Author: RomainLanz

src/queue_config_resolver.ts

@@ -0,0 +1,98 @@
+import type {
+  Duration,
+  JobOptions,
+  QueueConfig,
+  QueueManagerConfig,
+  RetryConfig,
+} from './types/main.js'
+
+/**
+ * Resolve effective queue/job runtime configuration from the initialized
+ * queue config.
+ *
+ * This keeps merge rules in one place without coupling execution code to the
+ * full `QueueManager` lifecycle and adapter concerns.
+ */
+export class QueueConfigResolver {
+  readonly #globalRetryConfig?: RetryConfig
+  readonly #globalJobOptions?: JobOptions
+  readonly #queueConfigs: Map<string, QueueConfig>
+  readonly #workerTimeout?: Duration
+
+  /**
+   * Create a resolver from the queue manager config.
+   */
+  static from(config: QueueManagerConfig): QueueConfigResolver {
+    return new QueueConfigResolver({
+      globalRetryConfig: config.retry,
+      globalJobOptions: config.defaultJobOptions,
+      queueConfigs: new Map(Object.entries(config.queues || {}) as [string, QueueConfig][]),
+      workerTimeout: config.worker?.timeout,
+    })
+  }
+
+  /**
+   * Create a resolver from already-materialized config fragments.
+   */
+  constructor({
+    globalRetryConfig,
+    globalJobOptions,
+    queueConfigs,
+    workerTimeout,
+  }: {
+    globalRetryConfig?: RetryConfig
+    globalJobOptions?: JobOptions
+    queueConfigs?: Map<string, QueueConfig>
+    workerTimeout?: Duration
+  }) {
+    this.#globalRetryConfig = globalRetryConfig
+    this.#globalJobOptions = globalJobOptions
+    this.#queueConfigs = queueConfigs ?? new Map()
+    this.#workerTimeout = workerTimeout
+  }
+
+  /**
+   * Resolve the retry policy for a job using priority: job > queue > global.
+   */
+  resolveRetryConfig(queue: string, jobRetryConfig?: RetryConfig): RetryConfig {
+    const queueConfig = this.#queueConfigs.get(queue)
+    const queueRetryConfig = queueConfig?.retry || {}
+
+    const maxRetries =
+      jobRetryConfig?.maxRetries ??
+      queueRetryConfig.maxRetries ??
+      this.#globalRetryConfig?.maxRetries ??
+      0
+
+    const backoff =
+      jobRetryConfig?.backoff || queueRetryConfig.backoff || this.#globalRetryConfig?.backoff
+
+    return { maxRetries, backoff }
+  }
+
+  /**
+   * Resolve effective retention options using priority: job > queue > global.
+   */
+  resolveJobOptions(queue: string, jobOptions?: JobOptions): JobOptions {
+    const queueConfig = this.#queueConfigs.get(queue)
+    const queueJobOptions = queueConfig?.defaultJobOptions
+
+    return {
+      removeOnComplete:
+        jobOptions?.removeOnComplete ??
+        queueJobOptions?.removeOnComplete ??
+        this.#globalJobOptions?.removeOnComplete,
+      removeOnFail:
+        jobOptions?.removeOnFail ??
+        queueJobOptions?.removeOnFail ??
+        this.#globalJobOptions?.removeOnFail,
+    }
+  }
+
+  /**
+   * Return the configured default worker timeout.
+   */
+  getWorkerTimeout(): Duration | undefined {
+    return this.#workerTimeout
+  }
+}

src/queue_manager.ts

@@ -3,25 +3,17 @@ import debug from './debug.js'
 import { Locator } from './locator.js'
 import { consoleLogger, type Logger } from './logger.js'
 import { FakeAdapter } from './drivers/fake_adapter.js'
+import { QueueConfigResolver } from './queue_config_resolver.js'
 import type { Adapter } from './contracts/adapter.js'
-import type {
-  AdapterFactory,
-  JobFactory,
-  JobOptions,
-  QueueConfig,
-  QueueManagerConfig,
-  RetryConfig,
-} from './types/main.js'
+import type { AdapterFactory, JobFactory, QueueManagerConfig } from './types/main.js'
 
 type QueueManagerFakeState = {
   defaultAdapter: string
   adapters: Record<string, AdapterFactory>
   adapterInstances: Map<string, Adapter>
-  globalRetryConfig?: RetryConfig
-  globalJobOptions?: JobOptions
-  queueConfigs: Map<string, QueueConfig>
   logger: Logger
   jobFactory?: JobFactory
+  configResolver: QueueConfigResolver
   fakeAdapter: FakeAdapter
 }
 
@@ -61,11 +53,9 @@ class QueueManagerSingleton {
   #defaultAdapter!: string
   #adapters: Record<string, AdapterFactory> = {}
   #adapterInstances: Map<string, Adapter> = new Map()
-  #globalRetryConfig?: RetryConfig
-  #globalJobOptions?: JobOptions
-  #queueConfigs: Map<string, QueueConfig> = new Map()
   #logger: Logger = consoleLogger
   #jobFactory?: JobFactory
+  #configResolver: QueueConfigResolver = new QueueConfigResolver({})
   #fakeState?: QueueManagerFakeState
 
   /**
@@ -101,16 +91,9 @@ class QueueManagerSingleton {
 
     this.#defaultAdapter = config.default
     this.#adapters = config.adapters
-    this.#globalRetryConfig = config.retry
-    this.#globalJobOptions = config.defaultJobOptions
     this.#logger = config.logger ?? consoleLogger
     this.#jobFactory = config.jobFactory
-
-    if (config.queues) {
-      for (const [queue, queueConfig] of Object.entries(config.queues)) {
-        this.#queueConfigs.set(queue, queueConfig as QueueConfig)
-      }
-    }
+    this.#configResolver = QueueConfigResolver.from(config)
 
     if (config.locations && config.locations.length > 0) {
       const registered = await Locator.registerFromGlob(config.locations)
@@ -216,11 +199,9 @@ class QueueManagerSingleton {
       defaultAdapter: this.#defaultAdapter,
       adapters: this.#adapters,
       adapterInstances: this.#adapterInstances,
-      globalRetryConfig: this.#globalRetryConfig,
-      globalJobOptions: this.#globalJobOptions,
-      queueConfigs: this.#queueConfigs,
       logger: this.#logger,
       jobFactory: this.#jobFactory,
+      configResolver: this.#configResolver,
       fakeAdapter,
     }
 
@@ -257,44 +238,9 @@ class QueueManagerSingleton {
     this.#defaultAdapter = state.defaultAdapter
     this.#adapters = state.adapters
     this.#adapterInstances = state.adapterInstances
-    this.#globalRetryConfig = state.globalRetryConfig
-    this.#globalJobOptions = state.globalJobOptions
-    this.#queueConfigs = state.queueConfigs
     this.#logger = state.logger
     this.#jobFactory = state.jobFactory
-  }
-
-  /**
-   * Get the merged retry configuration for a job.
-   *
-   * Configuration is merged with priority: job > queue > global.
-   * This allows specific jobs or queues to override global defaults.
-   *
-   * @param queue - The queue name
-   * @param jobRetryConfig - Optional job-level retry config
-   * @returns The merged retry configuration
-   *
-   * @example
-   * ```typescript
-   * // Global: maxRetries=3, Queue: maxRetries=5, Job: maxRetries=1
-   * // Result: maxRetries=1 (job wins)
-   * const config = QueueManager.getMergedRetryConfig('emails', { maxRetries: 1 })
-   * ```
-   */
-  getMergedRetryConfig(queue: string, jobRetryConfig?: RetryConfig): RetryConfig {
-    const queueConfig = this.#queueConfigs.get(queue)
-    const queueRetryConfig = queueConfig?.retry || {}
-
-    let maxRetries =
-      jobRetryConfig?.maxRetries ??
-      queueRetryConfig.maxRetries ??
-      this.#globalRetryConfig?.maxRetries ??
-      0
-
-    let backoff =
-      jobRetryConfig?.backoff || queueRetryConfig.backoff || this.#globalRetryConfig?.backoff
-
-    return { maxRetries, backoff }
+    this.#configResolver = state.configResolver
   }
 
   /**
@@ -307,22 +253,14 @@ class QueueManagerSingleton {
   }
 
   /**
-   * Get the merged job options for a job (priority: job > queue > global).
+   * Get the resolver responsible for effective queue/job runtime config.
    */
-  getMergedJobOptions(queue: string, jobOptions?: JobOptions): JobOptions {
-    const queueConfig = this.#queueConfigs.get(queue)
-    const queueJobOptions = queueConfig?.defaultJobOptions
-
-    return {
-      removeOnComplete:
-        jobOptions?.removeOnComplete ??
-        queueJobOptions?.removeOnComplete ??
-        this.#globalJobOptions?.removeOnComplete,
-      removeOnFail:
-        jobOptions?.removeOnFail ??
-        queueJobOptions?.removeOnFail ??
-        this.#globalJobOptions?.removeOnFail,
+  getConfigResolver(): QueueConfigResolver {
+    if (!this.#initialized) {
+      throw new errors.E_QUEUE_NOT_INITIALIZED()
     }
+
+    return this.#configResolver
   }
 
   #validateConfig(config: QueueManagerConfig): void {
@@ -376,6 +314,7 @@ class QueueManagerSingleton {
 
     this.#adapterInstances.clear()
     this.#initialized = false
+    this.#configResolver = new QueueConfigResolver({})
     this.#fakeState = undefined
   }
 }

src/worker.ts

@@ -331,7 +331,8 @@ export class Worker {
     debug('worker %s: executing job %s (%s)', this.#id, job.id, job.name)
 
     const { instance, options, timeout, context, payload } = await this.#initJob(job, queue)
-    const retention = QueueManager.getMergedJobOptions(queue, options)
+    const configResolver = QueueManager.getConfigResolver()
+    const retention = configResolver.resolveJobOptions(queue, options)
 
     try {
       await this.#executeWithTimeout(instance, payload, context, timeout)
@@ -349,7 +350,7 @@ export class Worker {
         return
       }
 
-      const mergedConfig = QueueManager.getMergedRetryConfig(queue, options.retry)
+      const mergedConfig = configResolver.resolveRetryConfig(queue, options.retry)
 
       if (typeof mergedConfig.maxRetries === 'undefined' || mergedConfig.maxRetries <= 0) {
         debug('worker %s: job %s has no retries configured, marking as failed', this.#id, job.id)
@@ -417,7 +418,7 @@ export class Worker {
       return { instance, options, timeout, context, payload: job.payload }
     } catch (error) {
       debug('worker %s: failed to initialize job %s (%s)', this.#id, job.id, job.name)
-      const retention = QueueManager.getMergedJobOptions(queue)
+      const retention = QueueManager.getConfigResolver().resolveJobOptions(queue)
       await this.#adapter.failJob(job.id, queue, error as Error, retention.removeOnFail)
       throw error
     }

tests/queue_config_resolver.spec.ts

@@ -0,0 +1,93 @@
+import { test } from '@japa/runner'
+import { exponentialBackoff } from '../src/strategies/backoff_strategy.js'
+import { QueueConfigResolver } from '../src/queue_config_resolver.js'
+
+test.group('QueueConfigResolver', () => {
+  test('should merge retry configurations correctly (global)', ({ assert }) => {
+    const resolver = QueueConfigResolver.from({
+      default: 'sync',
+      adapters: { sync: () => ({}) as any },
+      retry: { maxRetries: 5, backoff: exponentialBackoff() },
+    })
+
+    let config = resolver.resolveRetryConfig('default')
+    assert.equal(config.maxRetries, 5)
+  })
+
+  test('should merge retry configurations correctly (queue)', ({ assert }) => {
+    const resolver = QueueConfigResolver.from({
+      default: 'sync',
+      adapters: { sync: () => ({}) as any },
+      retry: { maxRetries: 5, backoff: exponentialBackoff() },
+      queues: {
+        email: { retry: { maxRetries: 3 } },
+      },
+    })
+
+    let config = resolver.resolveRetryConfig('email')
+    assert.equal(config.maxRetries, 3)
+  })
+
+  test('should merge retry configurations correctly (job)', ({ assert }) => {
+    const resolver = QueueConfigResolver.from({
+      default: 'sync',
+      adapters: { sync: () => ({}) as any },
+      retry: { maxRetries: 5, backoff: exponentialBackoff() },
+      queues: {
+        email: { retry: { maxRetries: 3 } },
+      },
+    })
+
+    let config = resolver.resolveRetryConfig('email', { maxRetries: 2 })
+    assert.equal(config.maxRetries, 2)
+  })
+
+  test('should respect maxRetries: 0 from job config over global/queue config', ({ assert }) => {
+    const resolver = QueueConfigResolver.from({
+      default: 'sync',
+      adapters: { sync: () => ({}) as any },
+      retry: { maxRetries: 5, backoff: exponentialBackoff() },
+      queues: {
+        email: { retry: { maxRetries: 3 } },
+      },
+    })
+
+    let config = resolver.resolveRetryConfig('email', { maxRetries: 0 })
+    assert.equal(config.maxRetries, 0)
+  })
+
+  test('should resolve job retention options with correct precedence', ({ assert }) => {
+    const resolver = QueueConfigResolver.from({
+      default: 'sync',
+      adapters: { sync: () => ({}) as any },
+      defaultJobOptions: { removeOnFail: { age: '7d' } },
+      queues: {
+        email: {
+          defaultJobOptions: {
+            removeOnFail: { age: '3d' },
+            removeOnComplete: { count: 50 },
+          },
+        },
+      },
+    })
+
+    const resolved = resolver.resolveJobOptions('email', {
+      removeOnComplete: false,
+    })
+
+    assert.deepEqual(resolved, {
+      removeOnComplete: false,
+      removeOnFail: { age: '3d' },
+    })
+  })
+
+  test('should expose configured worker timeout', ({ assert }) => {
+    const resolver = QueueConfigResolver.from({
+      default: 'sync',
+      adapters: { sync: () => ({}) as any },
+      worker: { timeout: '30s' },
+    })
+
+    assert.equal(resolver.getWorkerTimeout(), '30s')
+  })
+})