{{ message }}
Repository navigation
Expand file tree
/
Copy pathqueue_manager.ts
More file actions
510 lines (446 loc) · 14.9 KB
/
Copy pathqueue_manager.ts
File metadata and controls
510 lines (446 loc) · 14.9 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
import * as errors from './exceptions.js'
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 { JobExecutionRuntime } from './job_runtime.js'
import { parseTimeout } from './utils.js'
import type { Adapter } from './contracts/adapter.js'
import type { AdapterFactory, JobFactory, QueueManagerConfig } from './types/main.js'
const noopInternalOperationWrapper: NonNullable<
QueueManagerConfig['internalOperationWrapper']
> = async (fn) => fn()
const noopExecutionWrapper: NonNullable<QueueManagerConfig['executionWrapper']> = async (fn) => fn()
type QueueManagerFakeState = {
defaultAdapter: string
adapters: Record<string, AdapterFactory>
adapterInstances: Map<string, Adapter>
logger: Logger
jobFactory?: JobFactory
internalOperationWrapper?: QueueManagerConfig['internalOperationWrapper']
executionWrapper?: QueueManagerConfig['executionWrapper']
configResolver: QueueConfigResolver
locations: string[]
hotReload: boolean
fakeAdapter: FakeAdapter
}
/**
* Central configuration and adapter management for the queue system.
*
* The QueueManager is responsible for:
* - Initializing adapters and job registration
* - Providing adapter instances to workers and dispatchers
* - Managing retry configuration across global, queue, and job levels
*
* @example
* ```typescript
* import { QueueManager, exponentialBackoff } from '@boringnode/queue'
* import { redis } from '@boringnode/queue/drivers/redis_adapter'
*
* await QueueManager.init({
* default: 'redis',
* adapters: {
* redis: redis({ host: 'localhost' }),
* },
* locations: ['./jobs/**
* The Worker that holds the QueueManager configuration and its Adapters,
* from `worker.start()` until its stop completes. Module state, not exported
* by the package: only the Worker reserves and releases it.
*/
const workerReservation: { owner?: object; initializing?: object } = {}
/** Reserve QueueManager for a starting Worker: `init()` and `destroy()` reject meanwhile. */
export function reserveQueueManager(worker: object): void {
workerReservation.owner = worker
}
export function releaseQueueManager(worker: object): void {
if (workerReservation.owner === worker) workerReservation.owner = undefined
}
/**
* Initialize QueueManager on behalf of the Worker that reserved it. `init()`
* checks the reservation before its first await, so the exemption only
* covers this call, not a concurrent `init()` from the application.
*/
export function initQueueManagerForWorker(worker: object, config: QueueManagerConfig) {
workerReservation.initializing = worker
try {
return QueueManager.init(config)
} finally {
workerReservation.initializing = undefined
}
}
/**\/*.js'],
* retry: {
* maxRetries: 3,
* backoff: exponentialBackoff(),
* },
* })
*
* // Get the default adapter
* const adapter = QueueManager.use()
*
* // Clean up when done
* await QueueManager.destroy()
* ```
*/
class QueueManagerSingleton {
#initialized = false
#defaultAdapter!: string
#adapters: Record<string, AdapterFactory> = {}
#adapterInstances: Map<string, Adapter> = new Map()
#logger: Logger = consoleLogger
#jobFactory?: JobFactory
#internalOperationWrapper?: QueueManagerConfig['internalOperationWrapper']
#executionWrapper?: QueueManagerConfig['executionWrapper']
#configResolver: QueueConfigResolver = new QueueConfigResolver({})
#locations: string[] = []
#hotReload = false
#fakeState?: QueueManagerFakeState
/**
* Initialize the queue system with the given configuration.
*
* This must be called before using the queue system. It:
* - Validates the configuration
* - Registers adapters
* - Auto-discovers and registers job classes from `locations`
*
* @param config - The queue configuration
* @returns This instance for chaining
* @throws {E_CONFIGURATION_ERROR} If the configuration is invalid
*
* @example
* ```typescript
* await QueueManager.init({
* default: 'redis',
* adapters: {
* redis: redis(),
* postgres: knex(pgConfig),
* },
* locations: ['./jobs/**\/*.js'],
* })
* ```
*/
async init(config: QueueManagerConfig) {
debug('initializing queue manager with config: %O', config)
// Synchronous, before the first await: see initQueueManagerForWorker().
if (workerReservation.owner && workerReservation.owner !== workerReservation.initializing) {
throw new errors.E_INIT_WHILE_WORKER_RUNNING()
}
this.#validateConfig(config)
await this.#cleanupBeforeReinitialization()
this.#defaultAdapter = config.default
this.#adapters = config.adapters
this.#logger = config.logger ?? consoleLogger
this.#jobFactory = config.jobFactory
this.#internalOperationWrapper = config.internalOperationWrapper
this.#executionWrapper = config.executionWrapper
this.#configResolver = QueueConfigResolver.from(config)
this.#locations = config.locations ?? []
this.#hotReload = config.hotReload ?? false
if (config.autoLoadJobs ?? true) {
await this.loadJobs()
}
this.#initialized = true
return this
}
/**
* Load and register job classes from configured or explicit locations.
*
* This low-level API is useful for framework integrations that need to
* register jobs at a precise moment in their command lifecycle.
*
* @param locations - Optional glob patterns. Defaults to the configured locations.
* @returns Number of jobs successfully registered.
*
* @example
* ```typescript
* await QueueManager.init(config)
* await QueueManager.loadJobs()
*
* // Or with explicit locations
* await QueueManager.loadJobs(['./app/jobs/**\/*.js'])
* ```
*/
async loadJobs(locations: string[] = this.#locations): Promise<number> {
if (locations.length === 0) {
return 0
}
return Locator.registerFromGlob(locations, {
hotReload: this.#hotReload,
logger: this.#logger,
})
}
/**
* Destroy any materialized adapters from the current configuration before
* replacing it with a new one.
*/
async #cleanupBeforeReinitialization() {
const destroyedAdapters = new Set<Adapter>()
await this.#destroyAdapters(this.#adapterInstances, destroyedAdapters)
if (this.#fakeState) {
await this.#destroyAdapter('fake', this.#fakeState.fakeAdapter, destroyedAdapters)
await this.#destroyAdapters(this.#fakeState.adapterInstances, destroyedAdapters)
this.#fakeState = undefined
}
this.#adapterInstances.clear()
}
/**
* Destroy a collection of adapters while avoiding double-destroying the same
* instance through multiple references.
*/
async #destroyAdapters(adapters: Iterable<[string, Adapter]>, destroyedAdapters: Set<Adapter>) {
for (const [name, adapter] of adapters) {
await this.#destroyAdapter(name, adapter, destroyedAdapters)
}
}
/**
* Destroy a single adapter once for the current cleanup pass.
*/
async #destroyAdapter(name: string, adapter: Adapter, destroyedAdapters: Set<Adapter>) {
if (destroyedAdapters.has(adapter)) {
return
}
destroyedAdapters.add(adapter)
debug('destroying adapter "%s" before reinitialization', name)
await adapter.destroy()
}
/**
* Get an adapter instance by name.
*
* Adapter instances are cached and reused. If no name is provided,
* the default adapter is returned.
*
* @param adapter - Adapter name (optional, defaults to the default adapter)
* @returns The adapter instance
* @throws {E_QUEUE_NOT_INITIALIZED} If `init()` hasn't been called
* @throws {E_CONFIGURATION_ERROR} If the adapter is not registered
* @throws {E_ADAPTER_INIT_ERROR} If the adapter factory throws
*
* @example
* ```typescript
* // Get default adapter
* const adapter = QueueManager.use()
*
* // Get specific adapter
* const redisAdapter = QueueManager.use('redis')
* ```
*/
use(adapter?: string): Adapter {
if (!this.#initialized) {
throw new errors.E_QUEUE_NOT_INITIALIZED()
}
if (!adapter) {
adapter = this.#defaultAdapter
}
// Return cached instance if exists
const cached = this.#adapterInstances.get(adapter)
if (cached) {
return cached
}
const adapterFactory = this.#adapters[adapter]
if (!adapterFactory) {
throw new errors.E_CONFIGURATION_ERROR([`Adapter "${adapter}" is not registered`])
}
debug('using adapter "%s"', adapter)
try {
const instance = adapterFactory({ logger: this.#logger })
this.#adapterInstances.set(adapter, instance)
return instance
} catch (error) {
const message = error instanceof Error ? error.message : String(error)
throw new errors.E_ADAPTER_INIT_ERROR([adapter, message], { cause: error })
}
}
/**
* Replace all adapters with a fake adapter for testing.
*
* The fake adapter records pushed jobs and exposes assertion helpers.
* Use the `using` keyword to automatically restore the previous
* configuration when the variable goes out of scope, or call
* `restore()` manually.
*
* @example
* ```typescript
* using fake = QueueManager.fake()
*
* await SendEmailJob.dispatch({ to: 'user@example.com' })
*
* fake.assertPushed(SendEmailJob)
* // Automatically restored at end of scope
* ```
*/
fake(): FakeAdapter {
if (!this.#initialized) {
throw new errors.E_QUEUE_NOT_INITIALIZED()
}
if (this.#fakeState) {
return this.#fakeState.fakeAdapter
}
const fakeAdapter = new FakeAdapter()
fakeAdapter.onDispose(() => this.restore())
this.#fakeState = {
defaultAdapter: this.#defaultAdapter,
adapters: this.#adapters,
adapterInstances: this.#adapterInstances,
logger: this.#logger,
jobFactory: this.#jobFactory,
internalOperationWrapper: this.#internalOperationWrapper,
executionWrapper: this.#executionWrapper,
configResolver: this.#configResolver,
locations: this.#locations,
hotReload: this.#hotReload,
fakeAdapter,
}
const fakeFactory = () => fakeAdapter
const nextAdapters: Record<string, AdapterFactory> = {}
for (const name of Object.keys(this.#fakeState.adapters)) {
nextAdapters[name] = fakeFactory
}
this.#adapters = nextAdapters
this.#adapterInstances = new Map()
return fakeAdapter
}
/**
* Restore adapters after calling `fake()`.
*/
restore(): void {
if (!this.#fakeState) {
return
}
void this.#fakeState.fakeAdapter.destroy()
for (const adapter of this.#adapterInstances.values()) {
void adapter.destroy()
}
const state = this.#fakeState
this.#fakeState = undefined
this.#defaultAdapter = state.defaultAdapter
this.#adapters = state.adapters
this.#adapterInstances = state.adapterInstances
this.#logger = state.logger
this.#jobFactory = state.jobFactory
this.#internalOperationWrapper = state.internalOperationWrapper
this.#executionWrapper = state.executionWrapper
this.#configResolver = state.configResolver
this.#locations = state.locations
this.#hotReload = state.hotReload
}
/**
* Get the configured job factory for custom instantiation.
*
* @returns The job factory function, or undefined if not configured
*/
getJobFactory(): JobFactory | undefined {
return this.#jobFactory
}
/**
* Whether the queue manager has been initialized.
*/
isInitialized(): boolean {
return this.#initialized
}
/**
* Get the configured logger used by the queue runtime.
*/
getLogger(): Logger {
return this.#logger
}
/**
* Get the configured internal operation wrapper.
*/
getInternalOperationWrapper() {
return this.#internalOperationWrapper ?? noopInternalOperationWrapper
}
/**
* Get the configured execution wrapper.
*/
getExecutionWrapper() {
return this.#executionWrapper ?? noopExecutionWrapper
}
/**
* Get the resolver responsible for effective queue/job runtime config.
*/
getConfigResolver(): QueueConfigResolver {
if (!this.#initialized) {
throw new errors.E_QUEUE_NOT_INITIALIZED()
}
return this.#configResolver
}
/**
* Create the configured in-process Job execution runtime.
*/
getJobExecutionRuntime(): JobExecutionRuntime {
if (!this.#initialized) {
throw new errors.E_QUEUE_NOT_INITIALIZED()
}
return new JobExecutionRuntime({
resolveJob: (jobName) => Locator.resolve(jobName),
configResolver: this.#configResolver,
jobFactory: this.#jobFactory,
executionWrapper: this.#executionWrapper,
})
}
#validateConfig(config: QueueManagerConfig): void {
if (!config.adapters || Object.keys(config.adapters).length === 0) {
throw new errors.E_CONFIGURATION_ERROR(['At least one adapter must be configured'])
}
if (!config.default) {
throw new errors.E_CONFIGURATION_ERROR(['Default adapter must be specified'])
}
if (!config.adapters[config.default]) {
throw new errors.E_CONFIGURATION_ERROR([
`Default adapter "${config.default}" not found in adapters configuration`,
])
}
for (const [name, factory] of Object.entries(config.adapters)) {
if (typeof factory !== 'function') {
throw new errors.E_CONFIGURATION_ERROR([`Adapter "${name}" must be a factory function`])
}
}
// Fail at startup rather than when the first job runs.
parseTimeout(config.worker?.timeout)
parseTimeout(config.defaultJobOptions?.timeout)
for (const queueConfig of Object.values(config.queues ?? {})) {
parseTimeout(queueConfig.defaultJobOptions?.timeout)
}
}
/**
* Clean up all adapter instances and reset state.
*
* Call this when shutting down the application or when
* you need to reinitialize with a new configuration.
* Throws while a Worker runs in this process: stop it first.
*
* @example
* ```typescript
* // On application shutdown
* await QueueManager.destroy()
* ```
*/
async destroy() {
// Synchronous, before the first await, like init().
if (workerReservation.owner) {
throw new errors.E_DESTROY_WHILE_WORKER_RUNNING()
}
for (const [name, adapter] of this.#adapterInstances) {
debug('destroying adapter "%s"', name)
await adapter.destroy()
}
if (this.#fakeState) {
await this.#fakeState.fakeAdapter.destroy()
for (const [name, adapter] of this.#fakeState.adapterInstances) {
debug('destroying adapter "%s"', name)
await adapter.destroy()
}
}
this.#adapterInstances.clear()
this.#initialized = false
this.#internalOperationWrapper = undefined
this.#executionWrapper = undefined
this.#configResolver = new QueueConfigResolver({})
this.#locations = []
this.#hotReload = false
this.#fakeState = undefined
}
}
/** Global queue manager singleton */
export const QueueManager = new QueueManagerSingleton()
You can’t perform that action at this time.