16127615119

Committed 07 Jul 2025 08:51PM UTC coverage: 94.475% (-5.5%) from 100.0%

Build # 16127615119

Build Type

Pull #495

github

Committed by

web-flow

Commit Message

Merge ec219eae0 into 6d4c88633

Pull Request Pull Request #495: v11

Run Details

368 of 430 branches covered (85.58%)

304 of 305 new or added lines in 8 files covered. (99.67%)

49 existing lines in 4 files now uncovered.

855 of 905 relevant lines covered (94.48%)

103.5 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.65

/src/manager.js

const assert = require('node:assert')
const EventEmitter = require('node:events')
const { randomUUID } = require('node:crypto')
const { serializeError: stringify } = require('serialize-error')
const { delay, resolveWithinSeconds } = require('./tools')
const Attorney = require('./attorney')
const Worker = require('./worker')
const plans = require('./plans')

const { QUEUES: TIMEKEEPER_QUEUES } = require('./timekeeper')
const { QUEUE_POLICIES } = plans

const INTERNAL_QUEUES = Object.values(TIMEKEEPER_QUEUES).reduce((acc, i) => ({ ...acc, [i]: i }), {})

const events = {
  error: 'error',
  wip: 'wip'
}

class Manager extends EventEmitter {
  constructor (db, config) {
    super()

    this.config = config
    this.db = db
    this.wipTs = Date.now()
    this.workers = new Map()
    this.queues = null

    this.events = events
    this.functions = [
      this.complete,
      this.cancel,
      this.resume,
      this.retry,
      this.fail,
      this.fetch,
      this.work,
      this.offWork,
      this.notifyWorker,
      this.publish,
      this.subscribe,
      this.unsubscribe,
      this.insert,
      this.send,
      this.sendDebounced,
      this.sendThrottled,
      this.sendAfter,
      this.createQueue,
      this.updateQueue,
      this.deleteQueue,
      this.getQueueStats,
      this.getQueue,
      this.getQueues,
      this.deleteQueuedJobs,
      this.deleteStoredJobs,
      this.deleteAllJobs,
      this.deleteJob,
      this.getJobById
    ]
  }

  async start () {
    this.stopped = false
    this.queueCacheInterval = setInterval(() => this.onCacheQueues({ emit: true }), this.config.queueCacheIntervalSeconds * 1000)
    await this.onCacheQueues()
  }

  async onCacheQueues ({ emit = false } = {}) {
    try {
      assert(!this.config.__test__throw_queueCache, 'test error')
      const queues = await this.getQueues()
      this.queues = queues.reduce((acc, i) => { acc[i.name] = i; return acc }, {})
    } catch (error) {
      emit && this.emit(events.error, { ...error, message: error.message, stack: error.stack })
    }
  }

  async getQueueCache (name) {
    let queue = this.queues[name]

    if (queue) {
      return queue
    }

    queue = await this.getQueue(name)

    if (!queue) {
      throw new Error(`Queue ${name} does not exist`)
    }

    this.queues[name] = queue

    return queue
  }

  async stop () {
    this.stopped = true

    clearInterval(this.queueCacheInterval)

    for (const worker of this.workers.values()) {
      if (!INTERNAL_QUEUES[worker.name]) {
        await this.offWork(worker.name)
      }
    }
  }

  async failWip () {
    for (const worker of this.workers.values()) {
      const jobIds = worker.jobs.map(j => j.id)
      if (jobIds.length) {
        await this.fail(worker.name, jobIds, 'pg-boss shut down while active')
      }
    }
  }

  async work (name, ...args) {
    const { options, callback } = Attorney.checkWorkArgs(name, args)
    return await this.watch(name, options, callback)
  }

  addWorker (worker) {
    this.workers.set(worker.id, worker)
  }

  removeWorker (worker) {
    this.workers.delete(worker.id)
  }

  getWorkers () {
    return Array.from(this.workers.values())
  }

  emitWip (name) {
    if (!INTERNAL_QUEUES[name]) {
      const now = Date.now()

      if (now - this.wipTs > 2000) {
        this.emit(events.wip, this.getWipData())
        this.wipTs = now
      }
    }
  }

  getWipData (options = {}) {
    const { includeInternal = false } = options

    const data = this.getWorkers()
      .map(({
        id,
        name,
        options,
        state,
        jobs,
        createdOn,
        lastFetchedOn,
        lastJobStartedOn,
        lastJobEndedOn,
        lastError,
        lastErrorOn
      }) => ({
        id,
        name,
        options,
        state,
        count: jobs.length,
        createdOn,
        lastFetchedOn,
        lastJobStartedOn,
        lastJobEndedOn,
        lastError,
        lastErrorOn
      }))
      .filter(i => i.count > 0 && (!INTERNAL_QUEUES[i.name] || includeInternal))

    return data
  }

  async watch (name, options, callback) {
    if (this.stopped) {
      throw new Error('Workers are disabled. pg-boss is stopped')
    }

    const {
      pollingInterval: interval = this.config.pollingInterval,
      batchSize,
      includeMetadata = false,
      priority = true
    } = options

    const id = randomUUID({ disableEntropyCache: true })

    const fetch = () => this.fetch(name, { batchSize, includeMetadata, priority })

    const onFetch = async (jobs) => {
      if (!jobs.length) {
        return
      }

      if (this.config.__test__throw_worker) {
        throw new Error('__test__throw_worker')
      }

      this.emitWip(name)

      const maxExpiration = jobs.reduce((acc, i) => Math.max(acc, i.expireInSeconds), 0)
      const jobIds = jobs.map(job => job.id)

      try {
        const result = await resolveWithinSeconds(callback(jobs), maxExpiration, `handler execution exceeded ${maxExpiration}s`)
        this.complete(name, jobIds, jobIds.length === 1 ? result : undefined)
      } catch (err) {
        this.fail(name, jobIds, err)
      }

      this.emitWip(name)
    }

    const onError = error => {
      this.emit(events.error, { ...error, message: error.message, stack: error.stack, queue: name, worker: id })
    }

    const worker = new Worker({ id, name, options, interval, fetch, onFetch, onError })

    this.addWorker(worker)

    worker.start()

    return id
  }

  async offWork (value) {
    assert(value, 'Missing required argument')

    const query = (typeof value === 'string')
      ? { filter: i => i.name === value }
      : (typeof value === 'object' && value.id)
          ? { filter: i => i.id === value.id }
          : null

    assert(query, 'Invalid argument. Expected string or object: { id }')

    const workers = this.getWorkers().filter(i => query.filter(i) && !i.stopping && !i.stopped)

    if (workers.length === 0) {
      return
    }

    for (const worker of workers) {
      worker.stop()
    }

    setImmediate(async () => {
      while (!workers.every(w => w.stopped)) {
        await delay(1000)
      }

      for (const worker of workers) {
        this.removeWorker(worker)
      }
    })
  }

  notifyWorker (workerId) {
    if (this.workers.has(workerId)) {
      this.workers.get(workerId).notify()
    }
  }

  async subscribe (event, name) {
    assert(event, 'Missing required argument')
    assert(name, 'Missing required argument')
    const sql = plans.subscribe(this.config.schema)
    return await this.db.executeSql(sql, [event, name])
  }

  async unsubscribe (event, name) {
    assert(event, 'Missing required argument')
    assert(name, 'Missing required argument')
    const sql = plans.unsubscribe(this.config.schema)
    return await this.db.executeSql(sql, [event, name])
  }

  async publish (event, ...args) {
    assert(event, 'Missing required argument')
    const sql = plans.getQueuesForEvent(this.config.schema)
    const { rows } = await this.db.executeSql(sql, [event])

    await Promise.allSettled(rows.map(({ name }) => this.send(name, ...args)))
  }

  async send (...args) {
    const { name, data, options } = Attorney.checkSendArgs(args)

    return await this.createJob(name, data, options)
  }

  async sendAfter (name, data, options, after) {
    options = options ? { ...options } : {}
    options.startAfter = after

    const result = Attorney.checkSendArgs([name, data, options])

    return await this.createJob(result.name, result.data, result.options)
  }

  async sendThrottled (name, data, options, seconds, key) {
    options = options ? { ...options } : {}
    options.singletonSeconds = seconds
    options.singletonNextSlot = false
    options.singletonKey = key

    const result = Attorney.checkSendArgs([name, data, options])

    return await this.createJob(result.name, result.data, result.options)
  }

  async sendDebounced (name, data, options, seconds, key) {
    options = options ? { ...options } : {}
    options.singletonSeconds = seconds
    options.singletonNextSlot = true
    options.singletonKey = key

    const result = Attorney.checkSendArgs([name, data, options])

    return await this.createJob(result.name, result.data, result.options)
  }

  async createJob (name, data, options) {
    const singletonOffset = 0

    const {
      id = null,
      db: wrapper,
      priority,
      startAfter,
      singletonKey = null,
      singletonSeconds,
      singletonNextSlot,
      expireInSeconds,
      deleteAfterSeconds,
      keepUntil,
      retryLimit,
      retryDelay,
      retryBackoff,
      retryDelayMax
    } = options

    const job = {
      id,
      name,
      data,
      priority,
      startAfter,
      singletonKey,
      singletonSeconds,
      singletonOffset,
      expireInSeconds,
      deleteAfterSeconds,
      keepUntil,
      retryLimit,
      retryDelay,
      retryBackoff,
      retryDelayMax
    }

    const db = wrapper || this.db

    const { table } = await this.getQueueCache(name)

    const sql = plans.insertJobs(this.config.schema, { table, name, returnId: true })

    const { rows: try1 } = await db.executeSql(sql, [JSON.stringify([job])])

    if (try1.length === 1) {
      return try1[0].id
    }

    if (singletonNextSlot) {
      // delay starting by the offset to honor throttling config
      job.startAfter = this.getDebounceStartAfter(singletonSeconds, this.timekeeper.clockSkew)
      job.singletonOffset = singletonSeconds

      const { rows: try2 } = await db.executeSql(sql, [JSON.stringify([job])])

      if (try2.length === 1) {
        return try2[0].id
      }
    }

    return null
  }

  async insert (name, jobs, options = {}) {
    assert(Array.isArray(jobs), 'jobs argument should be an array')

    const { table } = await this.getQueueCache(name)

    const db = this.assertDb(options)

    const sql = plans.insertJobs(this.config.schema, { table, name, returnId: false })

    const { rows } = await db.executeSql(sql, [JSON.stringify(jobs)])

    return (rows.length) ? rows.map(i => i.id) : null
  }

  getDebounceStartAfter (singletonSeconds, clockOffset) {
    const debounceInterval = singletonSeconds * 1000

    const now = Date.now() + clockOffset

    const slot = Math.floor(now / debounceInterval) * debounceInterval

    // prevent startAfter=0 during debouncing
    let startAfter = (singletonSeconds - Math.floor((now - slot) / 1000)) || 1

    if (singletonSeconds > 1) {
      startAfter++
    }

    return startAfter
  }

  async fetch (name, options = {}) {
    Attorney.checkFetchArgs(name, options)

    const db = this.assertDb(options)

    const { table } = await this.getQueueCache(name)

    const sql = plans.fetchNextJob({ ...options, schema: this.config.schema, table, name, limit: options.batchSize })

    let result

    try {
      result = await db.executeSql(sql)
    } catch (err) {
      // errors from fetchquery should only be unique constraint violations
    }

    return result?.rows || []
  }

  mapCompletionIdArg (id, funcName) {
    const errorMessage = `${funcName}() requires an id`

    assert(id, errorMessage)

    const ids = Array.isArray(id) ? id : [id]

    assert(ids.length, errorMessage)

    return ids
  }

  mapCompletionDataArg (data) {
    if (data === null || typeof data === 'undefined' || typeof data === 'function') { return null }

    const result = (typeof data === 'object' && !Array.isArray(data))
      ? data
      : { value: data }

    return stringify(result)
  }

  mapCommandResponse (ids, result) {
    return {
      jobs: ids,
      requested: ids.length,
      affected: result && result.rows ? parseInt(result.rows[0].count) : 0
    }
  }

  async complete (name, id, data, options = {}) {
    Attorney.assertQueueName(name)
    const db = this.assertDb(options)
    const ids = this.mapCompletionIdArg(id, 'complete')
    const { table } = await this.getQueueCache(name)
    const sql = plans.completeJobs(this.config.schema, table)
    const result = await db.executeSql(sql, [name, ids, this.mapCompletionDataArg(data)])
    return this.mapCommandResponse(ids, result)
  }

  async fail (name, id, data, options = {}) {
    Attorney.assertQueueName(name)
    const db = this.assertDb(options)
    const ids = this.mapCompletionIdArg(id, 'fail')
    const { table } = await this.getQueueCache(name)
    const sql = plans.failJobsById(this.config.schema, table)
    const result = await db.executeSql(sql, [name, ids, this.mapCompletionDataArg(data)])
    return this.mapCommandResponse(ids, result)
  }

  async cancel (name, id, options = {}) {
    Attorney.assertQueueName(name)
    const db = this.assertDb(options)
    const ids = this.mapCompletionIdArg(id, 'cancel')
    const { table } = await this.getQueueCache(name)
    const sql = plans.cancelJobs(this.config.schema, table)
    const result = await db.executeSql(sql, [name, ids])
    return this.mapCommandResponse(ids, result)
  }

  async deleteJob (name, id, options = {}) {
    Attorney.assertQueueName(name)
    const db = this.assertDb(options)
    const ids = this.mapCompletionIdArg(id, 'deleteJob')
    const { table } = await this.getQueueCache(name)
    const sql = plans.deleteJobsById(this.config.schema, table)
    const result = await db.executeSql(sql, [name, ids])
    return this.mapCommandResponse(ids, result)
  }

  async resume (name, id, options = {}) {
    Attorney.assertQueueName(name)
    const db = this.assertDb(options)
    const ids = this.mapCompletionIdArg(id, 'resume')
    const { table } = await this.getQueueCache(name)
    const sql = plans.resumeJobs(this.config.schema, table)
    const result = await db.executeSql(sql, [name, ids])
    return this.mapCommandResponse(ids, result)
  }

  async retry (name, id, options = {}) {
    Attorney.assertQueueName(name)
    const db = options.db || this.db
    const ids = this.mapCompletionIdArg(id, 'retry')
    const { table } = await this.getQueueCache(name)
    const sql = plans.retryJobs(this.config.schema, table)
    const result = await db.executeSql(sql, [name, ids])
    return this.mapCommandResponse(ids, result)
  }

  async createQueue (name, options = {}) {
    name = name || options.name

    Attorney.assertQueueName(name)

    options.policy = options.policy || QUEUE_POLICIES.standard

    assert(options.policy in QUEUE_POLICIES, `${options.policy} is not a valid queue policy`)

    Attorney.validateQueueArgs(options)

    if (options.deadLetter) {
      Attorney.assertQueueName(options.deadLetter)
      assert.notStrictEqual(name, options.deadLetter, 'deadLetter cannot be itself')
      await this.getQueueCache(options.deadLetter)
    }

    const sql = plans.createQueue(this.config.schema, name, options)
    await this.db.executeSql(sql)
  }

  async getQueues (names) {
    if (names) {
      names = Array.isArray(names) ? names : [names]
      for (const name of names) {
        Attorney.assertQueueName(name)
      }
    }

    const sql = plans.getQueues(this.config.schema, names)
    const { rows } = await this.db.executeSql(sql)
    return rows
  }

  async updateQueue (name, options = {}) {
    Attorney.assertQueueName(name)

    assert(Object.keys(options).length > 0, 'no properties found to update')

    if ('policy' in options) {
      assert(options.policy in QUEUE_POLICIES, `${options.policy} is not a valid queue policy`)
    }

    Attorney.validateQueueArgs(options)

    const { deadLetter } = options

    if (deadLetter) {
      Attorney.assertQueueName(deadLetter)
      assert.notStrictEqual(name, deadLetter, 'deadLetter cannot be itself')
    }

    const sql = plans.updateQueue(this.config.schema, { deadLetter })
    await this.db.executeSql(sql, [name, options])
  }

  async getQueue (name) {
    Attorney.assertQueueName(name)

    const sql = plans.getQueues(this.config.schema, [name])
    const { rows } = await this.db.executeSql(sql)

    return rows[0] || null
  }

  async deleteQueue (name) {
    Attorney.assertQueueName(name)

    try {
      await this.getQueueCache(name)
      const sql = plans.deleteQueue(this.config.schema, name)
      await this.db.executeSql(sql)
    } catch {}
  }

  async deleteQueuedJobs (name) {
    Attorney.assertQueueName(name)
    const { table } = await this.getQueueCache(name)
    const sql = plans.deleteQueuedJobs(this.config.schema, table)
    await this.db.executeSql(sql, [name])
  }

  async deleteStoredJobs (name) {
    Attorney.assertQueueName(name)
    const { table } = await this.getQueueCache(name)
    const sql = plans.deleteStoredJobs(this.config.schema, table)
    await this.db.executeSql(sql, [name])
  }

  async deleteAllJobs (name) {
    Attorney.assertQueueName(name)
    const { table, partition } = await this.getQueueCache(name)

    if (partition) {
      const sql = plans.deleteAllJobs(this.config.schema, table)
      await this.db.executeSql(sql, [name])
    } else {
      const sql = plans.truncateTable(this.config.schema, table)
      await this.db.executeSql(sql)
    }
  }

  async getQueueStats (name) {
    Attorney.assertQueueName(name)

    const { table } = await this.getQueueCache(name)

    const sql = plans.getQueueStats(this.config.schema, table, [name])

    const { rows } = await this.db.executeSql(sql)

    return rows.at(0) || null
  }

  async getJobById (name, id, options = {}) {
    Attorney.assertQueueName(name)

    const db = this.assertDb(options)

    const { table } = await this.getQueueCache(name)

    const sql = plans.getJobById(this.config.schema, table)

    const result1 = await db.executeSql(sql, [name, id])

    if (result1?.rows?.length === 1) {
      return result1.rows[0]
    } else {
      return null
    }
  }

  assertDb (options) {
    if (options.db) {
      return options.db
    }

    assert(this.db._pgbdb && this.db.opened, 'Database connection is not opened')

    return this.db
  }
}

module.exports = Manager

1	const assert = require('node:assert')	4✔
2	const EventEmitter = require('node:events')	4✔
3	const { randomUUID } = require('node:crypto')	4✔
4	const { serializeError: stringify } = require('serialize-error')	4✔
5	const { delay, resolveWithinSeconds } = require('./tools')	4✔
6	const Attorney = require('./attorney')	4✔
7	const Worker = require('./worker')	4✔
8	const plans = require('./plans')	4✔
9
10	const { QUEUES: TIMEKEEPER_QUEUES } = require('./timekeeper')	4✔
11	const { QUEUE_POLICIES } = plans	4✔
12
13	const INTERNAL_QUEUES = Object.values(TIMEKEEPER_QUEUES).reduce((acc, i) => ({ ...acc, [i]: i }), {})	4✔
14
15	const events = {	4✔
16	error: 'error',
17	wip: 'wip'
18	}
19
20	class Manager extends EventEmitter {
21	constructor (db, config) {
22	super()	187✔
23
24	this.config = config	187✔
25	this.db = db	187✔
26	this.wipTs = Date.now()	187✔
27	this.workers = new Map()	187✔
28	this.queues = null	187✔
29
30	this.events = events	187✔
31	this.functions = [	187✔
32	this.complete,
33	this.cancel,
34	this.resume,
35	this.retry,
36	this.fail,
37	this.fetch,
38	this.work,
39	this.offWork,
40	this.notifyWorker,
41	this.publish,
42	this.subscribe,
43	this.unsubscribe,
44	this.insert,
45	this.send,
46	this.sendDebounced,
47	this.sendThrottled,
48	this.sendAfter,
49	this.createQueue,
50	this.updateQueue,
51	this.deleteQueue,
52	this.getQueueStats,
53	this.getQueue,
54	this.getQueues,
55	this.deleteQueuedJobs,
56	this.deleteStoredJobs,
57	this.deleteAllJobs,
58	this.deleteJob,
59	this.getJobById
60	]
61	}
62
63	async start () {
64	this.stopped = false	185✔
65	this.queueCacheInterval = setInterval(() => this.onCacheQueues({ emit: true }), this.config.queueCacheIntervalSeconds * 1000)	185✔
66	await this.onCacheQueues()	185✔
67	}
68
69	async onCacheQueues ({ emit = false } = {}) {	370✔
70	try {	187✔
71	assert(!this.config.__test__throw_queueCache, 'test error')	187✔
72	const queues = await this.getQueues()	184✔
73	this.queues = queues.reduce((acc, i) => { acc[i.name] = i; return acc }, {})	184✔
74	} catch (error) {
75	emit && this.emit(events.error, { ...error, message: error.message, stack: error.stack })	3✔
76	}
77	}
78
79	async getQueueCache (name) {
80	let queue = this.queues[name]	565✔
81
82	if (queue) {	564✔
83	return queue	413✔
84	}
85
86	queue = await this.getQueue(name)	151✔
87
88	if (!queue) {	151!
NEW 89	throw new Error(`Queue ${name} does not exist`)	×
90	}
91
92	this.queues[name] = queue	151✔
93
94	return queue	151✔
95	}
96
97	async stop () {
98	this.stopped = true	185✔
99
100	clearInterval(this.queueCacheInterval)	185✔
101
102	for (const worker of this.workers.values()) {	185✔
103	if (!INTERNAL_QUEUES[worker.name]) {	35✔
104	await this.offWork(worker.name)	25✔
105	}
106	}
107	}
108
109	async failWip () {
110	for (const worker of this.workers.values()) {	183✔
111	const jobIds = worker.jobs.map(j => j.id)	11✔
112	if (jobIds.length) {	11✔
113	await this.fail(worker.name, jobIds, 'pg-boss shut down while active')	1✔
114	}
115	}
116	}
117
118	async work (name, ...args) {
119	const { options, callback } = Attorney.checkWorkArgs(name, args)	40✔
120	return await this.watch(name, options, callback)	37✔
121	}
122
123	addWorker (worker) {
124	this.workers.set(worker.id, worker)	36✔
125	}
126
127	removeWorker (worker) {
128	this.workers.delete(worker.id)	35✔
129	}
130
131	getWorkers () {
132	return Array.from(this.workers.values())	222✔
133	}
134
135	emitWip (name) {
136	if (!INTERNAL_QUEUES[name]) {	104✔
137	const now = Date.now()	78✔
138
139	if (now - this.wipTs > 2000) {	78✔
140	this.emit(events.wip, this.getWipData())	16✔
141	this.wipTs = now	16✔
142	}
143	}
144	}
145
146	getWipData (options = {}) {	16✔
147	const { includeInternal = false } = options	185✔
148
149	const data = this.getWorkers()	185✔
150	.map(({
151	id,
152	name,
153	options,
154	state,
155	jobs,
156	createdOn,
157	lastFetchedOn,
158	lastJobStartedOn,
159	lastJobEndedOn,
160	lastError,
161	lastErrorOn
162	}) => ({	35✔
163	id,
164	name,
165	options,
166	state,
167	count: jobs.length,
168	createdOn,
169	lastFetchedOn,
170	lastJobStartedOn,
171	lastJobEndedOn,
172	lastError,
173	lastErrorOn
174	}))
175	.filter(i => i.count > 0 && (!INTERNAL_QUEUES[i.name] \|\| includeInternal))	35!
176
177	return data	185✔
178	}
179
180	async watch (name, options, callback) {
181	if (this.stopped) {	37✔
182	throw new Error('Workers are disabled. pg-boss is stopped')	1✔
183	}
184
185	const {
186	pollingInterval: interval = this.config.pollingInterval,	×
187	batchSize,
188	includeMetadata = false,	34✔
189	priority = true	36✔
190	} = options	36✔
191
192	const id = randomUUID({ disableEntropyCache: true })	36✔
193
194	const fetch = () => this.fetch(name, { batchSize, includeMetadata, priority })	111✔
195
196	const onFetch = async (jobs) => {	36✔
197	if (!jobs.length) {	111✔
198	return	58✔
199	}
200
201	if (this.config.__test__throw_worker) {	53✔
202	throw new Error('__test__throw_worker')	1✔
203	}
204
205	this.emitWip(name)	52✔
206
207	const maxExpiration = jobs.reduce((acc, i) => Math.max(acc, i.expireInSeconds), 0)	56✔
208	const jobIds = jobs.map(job => job.id)	56✔
209
210	try {	52✔
211	const result = await resolveWithinSeconds(callback(jobs), maxExpiration, `handler execution exceeded ${maxExpiration}s`)	52✔
212	this.complete(name, jobIds, jobIds.length === 1 ? result : undefined)	36✔
213	} catch (err) {
214	this.fail(name, jobIds, err)	16✔
215	}
216
217	this.emitWip(name)	52✔
218	}
219
220	const onError = error => {	36✔
221	this.emit(events.error, { ...error, message: error.message, stack: error.stack, queue: name, worker: id })	1✔
222	}
223
224	const worker = new Worker({ id, name, options, interval, fetch, onFetch, onError })	36✔
225
226	this.addWorker(worker)	36✔
227
228	worker.start()	36✔
229
230	return id	36✔
231	}
232
233	async offWork (value) {
234	assert(value, 'Missing required argument')	38✔
235
236	const query = (typeof value === 'string')	37✔
237	? { filter: i => i.name === value }	38✔
238	: (typeof value === 'object' && value.id)	3!
239	? { filter: i => i.id === value.id }	1✔
240	: null
241
242	assert(query, 'Invalid argument. Expected string or object: { id }')	37✔
243
244	const workers = this.getWorkers().filter(i => query.filter(i) && !i.stopping && !i.stopped)	39✔
245
246	if (workers.length === 0) {	37✔
247	return	2✔
248	}
249
250	for (const worker of workers) {	35✔
251	worker.stop()	36✔
252	}
253
254	setImmediate(async () => {	35✔
255	while (!workers.every(w => w.stopped)) {	47✔
256	await delay(1000)	12✔
257	}
258
259	for (const worker of workers) {	34✔
260	this.removeWorker(worker)	35✔
261	}
262	})
263	}
264
265	notifyWorker (workerId) {
266	if (this.workers.has(workerId)) {	1!
267	this.workers.get(workerId).notify()	1✔
268	}
269	}
270
271	async subscribe (event, name) {
272	assert(event, 'Missing required argument')	5✔
273	assert(name, 'Missing required argument')	5✔
274	const sql = plans.subscribe(this.config.schema)	5✔
275	return await this.db.executeSql(sql, [event, name])	5✔
276	}
277
278	async unsubscribe (event, name) {
279	assert(event, 'Missing required argument')	4✔
280	assert(name, 'Missing required argument')	3✔
281	const sql = plans.unsubscribe(this.config.schema)	2✔
282	return await this.db.executeSql(sql, [event, name])	2✔
283	}
284
285	async publish (event, ...args) {
286	assert(event, 'Missing required argument')	11✔
287	const sql = plans.getQueuesForEvent(this.config.schema)	10✔
288	const { rows } = await this.db.executeSql(sql, [event])	10✔
289
290	await Promise.allSettled(rows.map(({ name }) => this.send(name, ...args)))	10✔
291	}
292
293	async send (...args) {
294	const { name, data, options } = Attorney.checkSendArgs(args)	186✔
295
296	return await this.createJob(name, data, options)	183✔
297	}
298
299	async sendAfter (name, data, options, after) {
300	options = options ? { ...options } : {}	1!
301	options.startAfter = after	1✔
302
303	const result = Attorney.checkSendArgs([name, data, options])	1✔
304
305	return await this.createJob(result.name, result.data, result.options)	1✔
306	}
307
308	async sendThrottled (name, data, options, seconds, key) {
309	options = options ? { ...options } : {}	2!
310	options.singletonSeconds = seconds	2✔
311	options.singletonNextSlot = false	2✔
312	options.singletonKey = key	2✔
313
314	const result = Attorney.checkSendArgs([name, data, options])	2✔
315
316	return await this.createJob(result.name, result.data, result.options)	2✔
317	}
318
319	async sendDebounced (name, data, options, seconds, key) {
320	options = options ? { ...options } : {}	3!
321	options.singletonSeconds = seconds	3✔
322	options.singletonNextSlot = true	3✔
323	options.singletonKey = key	3✔
324
325	const result = Attorney.checkSendArgs([name, data, options])	3✔
326
327	return await this.createJob(result.name, result.data, result.options)	3✔
328	}
329
330	async createJob (name, data, options) {
331	const singletonOffset = 0	189✔
332
333	const {
334	id = null,	189✔
335	db: wrapper,
336	priority,
337	startAfter,
338	singletonKey = null,	176✔
339	singletonSeconds,
340	singletonNextSlot,
341	expireInSeconds,
342	deleteAfterSeconds,
343	keepUntil,
344	retryLimit,
345	retryDelay,
346	retryBackoff,
347	retryDelayMax
348	} = options	189✔
349
350	const job = {	189✔
351	id,
352	name,
353	data,
354	priority,
355	startAfter,
356	singletonKey,
357	singletonSeconds,
358	singletonOffset,
359	expireInSeconds,
360	deleteAfterSeconds,
361	keepUntil,
362	retryLimit,
363	retryDelay,
364	retryBackoff,
365	retryDelayMax
366	}
367
368	const db = wrapper \|\| this.db	189✔
369
370	const { table } = await this.getQueueCache(name)	189✔
371
372	const sql = plans.insertJobs(this.config.schema, { table, name, returnId: true })	188✔
373
374	const { rows: try1 } = await db.executeSql(sql, [JSON.stringify([job])])	188✔
375
376	if (try1.length === 1) {	188✔
377	return try1[0].id	175✔
378	}
379
380	if (singletonNextSlot) {	13✔
381	// delay starting by the offset to honor throttling config
382	job.startAfter = this.getDebounceStartAfter(singletonSeconds, this.timekeeper.clockSkew)	3✔
383	job.singletonOffset = singletonSeconds	3✔
384
385	const { rows: try2 } = await db.executeSql(sql, [JSON.stringify([job])])	3✔
386
387	if (try2.length === 1) {	3✔
388	return try2[0].id	2✔
389	}
390	}
391
392	return null	11✔
393	}
394
395	async insert (name, jobs, options = {}) {	20✔
396	assert(Array.isArray(jobs), 'jobs argument should be an array')	21✔
397
398	const { table } = await this.getQueueCache(name)	21✔
399
400	const db = this.assertDb(options)	21✔
401
402	const sql = plans.insertJobs(this.config.schema, { table, name, returnId: false })	21✔
403
404	const { rows } = await db.executeSql(sql, [JSON.stringify(jobs)])	21✔
405
406	return (rows.length) ? rows.map(i => i.id) : null	21!
407	}
408
409	getDebounceStartAfter (singletonSeconds, clockOffset) {
410	const debounceInterval = singletonSeconds * 1000	3✔
411
412	const now = Date.now() + clockOffset	3✔
413
414	const slot = Math.floor(now / debounceInterval) * debounceInterval	3✔
415
416	// prevent startAfter=0 during debouncing
417	let startAfter = (singletonSeconds - Math.floor((now - slot) / 1000)) \|\| 1	3!
418
419	if (singletonSeconds > 1) {	3!
420	startAfter++	3✔
421	}
422
423	return startAfter	3✔
424	}
425
426	async fetch (name, options = {}) {	79✔
427	Attorney.checkFetchArgs(name, options)	207✔
428
429	const db = this.assertDb(options)	206✔
430
431	const { table } = await this.getQueueCache(name)	206✔
432
433	const sql = plans.fetchNextJob({ ...options, schema: this.config.schema, table, name, limit: options.batchSize })	206✔
434
435	let result
436
437	try {	206✔
438	result = await db.executeSql(sql)	206✔
439	} catch (err) {
440	// errors from fetchquery should only be unique constraint violations
441	}
442
443	return result?.rows \|\| []	206✔
444	}
445
446	mapCompletionIdArg (id, funcName) {
447	const errorMessage = `${funcName}() requires an id`	92✔
448
449	assert(id, errorMessage)	92✔
450
451	const ids = Array.isArray(id) ? id : [id]	92✔
452
453	assert(ids.length, errorMessage)	92✔
454
455	return ids	92✔
456	}
457
458	mapCompletionDataArg (data) {
459	if (data === null \|\| typeof data === 'undefined' \|\| typeof data === 'function') { return null }	81✔
460
461	const result = (typeof data === 'object' && !Array.isArray(data))	31✔
462	? data
463	: { value: data }
464
465	return stringify(result)	31✔
466	}
467
468	mapCommandResponse (ids, result) {
469	return {	92✔
470	jobs: ids,
471	requested: ids.length,
472	affected: result && result.rows ? parseInt(result.rows[0].count) : 0	276!
473	}
474	}
475
476	async complete (name, id, data, options = {}) {	48✔
477	Attorney.assertQueueName(name)	49✔
478	const db = this.assertDb(options)	48✔
479	const ids = this.mapCompletionIdArg(id, 'complete')	48✔
480	const { table } = await this.getQueueCache(name)	48✔
481	const sql = plans.completeJobs(this.config.schema, table)	48✔
482	const result = await db.executeSql(sql, [name, ids, this.mapCompletionDataArg(data)])	48✔
483	return this.mapCommandResponse(ids, result)	48✔
484	}
485
486	async fail (name, id, data, options = {}) {	33✔
487	Attorney.assertQueueName(name)	34✔
488	const db = this.assertDb(options)	33✔
489	const ids = this.mapCompletionIdArg(id, 'fail')	33✔
490	const { table } = await this.getQueueCache(name)	33✔
491	const sql = plans.failJobsById(this.config.schema, table)	33✔
492	const result = await db.executeSql(sql, [name, ids, this.mapCompletionDataArg(data)])	33✔
493	return this.mapCommandResponse(ids, result)	33✔
494	}
495
496	async cancel (name, id, options = {}) {	5✔
497	Attorney.assertQueueName(name)	7✔
498	const db = this.assertDb(options)	6✔
499	const ids = this.mapCompletionIdArg(id, 'cancel')	6✔
500	const { table } = await this.getQueueCache(name)	6✔
501	const sql = plans.cancelJobs(this.config.schema, table)	6✔
502	const result = await db.executeSql(sql, [name, ids])	6✔
503	return this.mapCommandResponse(ids, result)	6✔
504	}
505
506	async deleteJob (name, id, options = {}) {	2✔
507	Attorney.assertQueueName(name)	2✔
508	const db = this.assertDb(options)	2✔
509	const ids = this.mapCompletionIdArg(id, 'deleteJob')	2✔
510	const { table } = await this.getQueueCache(name)	2✔
511	const sql = plans.deleteJobsById(this.config.schema, table)	2✔
512	const result = await db.executeSql(sql, [name, ids])	2✔
513	return this.mapCommandResponse(ids, result)	2✔
514	}
515
516	async resume (name, id, options = {}) {	2✔
517	Attorney.assertQueueName(name)	3✔
518	const db = this.assertDb(options)	2✔
519	const ids = this.mapCompletionIdArg(id, 'resume')	2✔
520	const { table } = await this.getQueueCache(name)	2✔
521	const sql = plans.resumeJobs(this.config.schema, table)	2✔
522	const result = await db.executeSql(sql, [name, ids])	2✔
523	return this.mapCommandResponse(ids, result)	2✔
524	}
525
526	async retry (name, id, options = {}) {	1✔
527	Attorney.assertQueueName(name)	1✔
528	const db = options.db \|\| this.db	1✔
529	const ids = this.mapCompletionIdArg(id, 'retry')	1✔
530	const { table } = await this.getQueueCache(name)	1✔
531	const sql = plans.retryJobs(this.config.schema, table)	1✔
532	const result = await db.executeSql(sql, [name, ids])	1✔
533	return this.mapCommandResponse(ids, result)	1✔
534	}
535
536	async createQueue (name, options = {}) {	156✔
537	name = name \|\| options.name	176!
538
539	Attorney.assertQueueName(name)	176✔
540
541	options.policy = options.policy \|\| QUEUE_POLICIES.standard	176✔
542
543	assert(options.policy in QUEUE_POLICIES, `${options.policy} is not a valid queue policy`)	176✔
544
545	Attorney.validateQueueArgs(options)	175✔
546
547	if (options.deadLetter) {	175✔
548	Attorney.assertQueueName(options.deadLetter)	4✔
549	assert.notStrictEqual(name, options.deadLetter, 'deadLetter cannot be itself')	4✔
550	await this.getQueueCache(options.deadLetter)	4✔
551	}
552
553	const sql = plans.createQueue(this.config.schema, name, options)	175✔
554	await this.db.executeSql(sql)	175✔
555	}
556
557	async getQueues (names) {
558	if (names) {	203✔
559	names = Array.isArray(names) ? names : [names]	7!
560	for (const name of names) {	7✔
561	Attorney.assertQueueName(name)	7✔
562	}
563	}
564
565	const sql = plans.getQueues(this.config.schema, names)	203✔
566	const { rows } = await this.db.executeSql(sql)	203✔
567	return rows	203✔
568	}
569
570	async updateQueue (name, options = {}) {	×
571	Attorney.assertQueueName(name)	3✔
572
573	assert(Object.keys(options).length > 0, 'no properties found to update')	3✔
574
575	if ('policy' in options) {	3✔
576	assert(options.policy in QUEUE_POLICIES, `${options.policy} is not a valid queue policy`)	1✔
577	}
578
579	Attorney.validateQueueArgs(options)	3✔
580
581	const { deadLetter } = options	3✔
582
583	if (deadLetter) {	3!
584	Attorney.assertQueueName(deadLetter)	3✔
585	assert.notStrictEqual(name, deadLetter, 'deadLetter cannot be itself')	3✔
586	}
587
588	const sql = plans.updateQueue(this.config.schema, { deadLetter })	3✔
589	await this.db.executeSql(sql, [name, options])	3✔
590	}
591
592	async getQueue (name) {
593	Attorney.assertQueueName(name)	159✔
594
595	const sql = plans.getQueues(this.config.schema, [name])	159✔
596	const { rows } = await this.db.executeSql(sql)	159✔
597
598	return rows[0] \|\| null	159✔
599	}
600
601	async deleteQueue (name) {
602	Attorney.assertQueueName(name)	5✔
603
604	try {	5✔
605	await this.getQueueCache(name)	5✔
606	const sql = plans.deleteQueue(this.config.schema, name)	5✔
607	await this.db.executeSql(sql)	5✔
608	} catch {}
609	}
610
611	async deleteQueuedJobs (name) {
612	Attorney.assertQueueName(name)	1✔
613	const { table } = await this.getQueueCache(name)	1✔
614	const sql = plans.deleteQueuedJobs(this.config.schema, table)	1✔
615	await this.db.executeSql(sql, [name])	1✔
616	}
617
618	async deleteStoredJobs (name) {
619	Attorney.assertQueueName(name)	1✔
620	const { table } = await this.getQueueCache(name)	1✔
621	const sql = plans.deleteStoredJobs(this.config.schema, table)	1✔
622	await this.db.executeSql(sql, [name])	1✔
623	}
624
625	async deleteAllJobs (name) {
626	Attorney.assertQueueName(name)	3✔
627	const { table, partition } = await this.getQueueCache(name)	3✔
628
629	if (partition) {	3✔
630	const sql = plans.deleteAllJobs(this.config.schema, table)	2✔
631	await this.db.executeSql(sql, [name])	2✔
632	} else {
633	const sql = plans.truncateTable(this.config.schema, table)	1✔
634	await this.db.executeSql(sql)	1✔
635	}
636	}
637
638	async getQueueStats (name) {
639	Attorney.assertQueueName(name)	2✔
640
641	const { table } = await this.getQueueCache(name)	2✔
642
643	const sql = plans.getQueueStats(this.config.schema, table, [name])	2✔
644
645	const { rows } = await this.db.executeSql(sql)	2✔
646
647	return rows.at(0) \|\| null	2!
648	}
649
650	async getJobById (name, id, options = {}) {	39✔
651	Attorney.assertQueueName(name)	41✔
652
653	const db = this.assertDb(options)	41✔
654
655	const { table } = await this.getQueueCache(name)	41✔
656
657	const sql = plans.getJobById(this.config.schema, table)	41✔
658
659	const result1 = await db.executeSql(sql, [name, id])	41✔
660
661	if (result1?.rows?.length === 1) {	41✔
662	return result1.rows[0]	36✔
663	} else {
664	return null	5✔
665	}
666	}
667
668	assertDb (options) {
669	if (options.db) {	359✔
670	return options.db	10✔
671	}
672
673	assert(this.db._pgbdb && this.db.opened, 'Database connection is not opened')	349✔
674
675	return this.db	349✔
676	}
677	}
678
679	module.exports = Manager	4✔

timgit / pg-boss / 16127615119

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous