17884078814

Committed 20 Sep 2025 07:42PM UTC coverage: 94.487% (-5.5%) from 100.0%

Build # 17884078814

Build Type

Pull #495

github

Committed by

web-flow

Commit Message

Merge efac4e87f into 60d830619

Pull Request Pull Request #495: v11

Run Details

372 of 433 branches covered (85.91%)

308 of 309 new or added lines in 8 files covered. (99.68%)

49 existing lines in 4 files now uncovered.

857 of 907 relevant lines covered (94.49%)

105.29 hits per line

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

99.66

/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`)
        await this.complete(name, jobIds, jobIds.length === 1 ? result : undefined)
      } catch (err) {
        await 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, singletonsActive } = await this.getQueueCache(name)

    options = {
      ...options,
      schema: this.config.schema,
      table,
      name,
      limit: options.batchSize,
      ignoreSingletons: singletonsActive
    }

    const sql = plans.fetchNextJob(options)

    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()	189✔
23
24	this.config = config	189✔
25	this.db = db	189✔
26	this.wipTs = Date.now()	189✔
27	this.workers = new Map()	189✔
28	this.queues = null	189✔
29
30	this.events = events	189✔
31	this.functions = [	189✔
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	187✔
65	this.queueCacheInterval = setInterval(() => this.onCacheQueues({ emit: true }), this.config.queueCacheIntervalSeconds * 1000)	187✔
66	await this.onCacheQueues()	187✔
67	}
68
69	async onCacheQueues ({ emit = false } = {}) {	374✔
70	try {	191✔
71	assert(!this.config.__test__throw_queueCache, 'test error')	191✔
72	const queues = await this.getQueues()	188✔
73	this.queues = queues.reduce((acc, i) => { acc[i.name] = i; return acc }, {})	188✔
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]	579✔
81
82	if (queue) {	578✔
83	return queue	425✔
84	}
85
86	queue = await this.getQueue(name)	153✔
87
88	if (!queue) {	153!
NEW 89	throw new Error(`Queue ${name} does not exist`)	×
90	}
91
92	this.queues[name] = queue	153✔
93
94	return queue	153✔
95	}
96
97	async stop () {
98	this.stopped = true	187✔
99
100	clearInterval(this.queueCacheInterval)	187✔
101
102	for (const worker of this.workers.values()) {	187✔
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()) {	185✔
111	const jobIds = worker.jobs.map(j => j.id)	16✔
112	if (jobIds.length) {	16✔
113	await this.fail(worker.name, jobIds, 'pg-boss shut down while active')	2✔
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())	227✔
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	190✔
148
149	const data = this.getWorkers()	190✔
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	}) => ({	43✔
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))	43✔
176
177	return data	190✔
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 })	112✔
195
196	const onFetch = async (jobs) => {	36✔
197	if (!jobs.length) {	112✔
198	return	59✔
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	await this.complete(name, jobIds, jobIds.length === 1 ? result : undefined)	36✔
213	} catch (err) {
214	await 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)) {	54✔
256	await delay(1000)	19✔
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)	192✔
295
296	return await this.createJob(name, data, options)	189✔
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	195✔
332
333	const {
334	id = null,	195✔
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	195✔
349
350	const job = {	195✔
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	195✔
369
370	const { table } = await this.getQueueCache(name)	195✔
371
372	const sql = plans.insertJobs(this.config.schema, { table, name, returnId: true })	194✔
373
374	const { rows: try1 } = await db.executeSql(sql, [JSON.stringify([job])])	194✔
375
376	if (try1.length === 1) {	194✔
377	return try1[0].id	182✔
378	}
379
380	if (singletonNextSlot) {	12✔
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	10✔
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 = {}) {	85✔
427	Attorney.checkFetchArgs(name, options)	214✔
428
429	const db = this.assertDb(options)	213✔
430
431	const { table, singletonsActive } = await this.getQueueCache(name)	213✔
432
433	options = {	213✔
434	...options,
435	schema: this.config.schema,
436	table,
437	name,
438	limit: options.batchSize,
439	ignoreSingletons: singletonsActive
440	}
441
442	const sql = plans.fetchNextJob(options)	213✔
443
444	let result
445
446	try {	213✔
447	result = await db.executeSql(sql)	213✔
448	} catch (err) {
449	// errors from fetchquery should only be unique constraint violations
450	}
451
452	return result?.rows \|\| []	213✔
453	}
454
455	mapCompletionIdArg (id, funcName) {
456	const errorMessage = `${funcName}() requires an id`	93✔
457
458	assert(id, errorMessage)	93✔
459
460	const ids = Array.isArray(id) ? id : [id]	93✔
461
462	assert(ids.length, errorMessage)	93✔
463
464	return ids	93✔
465	}
466
467	mapCompletionDataArg (data) {
468	if (data === null \|\| typeof data === 'undefined' \|\| typeof data === 'function') { return null }	82✔
469
470	const result = (typeof data === 'object' && !Array.isArray(data))	32✔
471	? data
472	: { value: data }
473
474	return stringify(result)	32✔
475	}
476
477	mapCommandResponse (ids, result) {
478	return {	93✔
479	jobs: ids,
480	requested: ids.length,
481	affected: result && result.rows ? parseInt(result.rows[0].count) : 0	279!
482	}
483	}
484
485	async complete (name, id, data, options = {}) {	48✔
486	Attorney.assertQueueName(name)	49✔
487	const db = this.assertDb(options)	48✔
488	const ids = this.mapCompletionIdArg(id, 'complete')	48✔
489	const { table } = await this.getQueueCache(name)	48✔
490	const sql = plans.completeJobs(this.config.schema, table)	48✔
491	const result = await db.executeSql(sql, [name, ids, this.mapCompletionDataArg(data)])	48✔
492	return this.mapCommandResponse(ids, result)	48✔
493	}
494
495	async fail (name, id, data, options = {}) {	34✔
496	Attorney.assertQueueName(name)	35✔
497	const db = this.assertDb(options)	34✔
498	const ids = this.mapCompletionIdArg(id, 'fail')	34✔
499	const { table } = await this.getQueueCache(name)	34✔
500	const sql = plans.failJobsById(this.config.schema, table)	34✔
501	const result = await db.executeSql(sql, [name, ids, this.mapCompletionDataArg(data)])	34✔
502	return this.mapCommandResponse(ids, result)	34✔
503	}
504
505	async cancel (name, id, options = {}) {	5✔
506	Attorney.assertQueueName(name)	7✔
507	const db = this.assertDb(options)	6✔
508	const ids = this.mapCompletionIdArg(id, 'cancel')	6✔
509	const { table } = await this.getQueueCache(name)	6✔
510	const sql = plans.cancelJobs(this.config.schema, table)	6✔
511	const result = await db.executeSql(sql, [name, ids])	6✔
512	return this.mapCommandResponse(ids, result)	6✔
513	}
514
515	async deleteJob (name, id, options = {}) {	2✔
516	Attorney.assertQueueName(name)	2✔
517	const db = this.assertDb(options)	2✔
518	const ids = this.mapCompletionIdArg(id, 'deleteJob')	2✔
519	const { table } = await this.getQueueCache(name)	2✔
520	const sql = plans.deleteJobsById(this.config.schema, table)	2✔
521	const result = await db.executeSql(sql, [name, ids])	2✔
522	return this.mapCommandResponse(ids, result)	2✔
523	}
524
525	async resume (name, id, options = {}) {	2✔
526	Attorney.assertQueueName(name)	3✔
527	const db = this.assertDb(options)	2✔
528	const ids = this.mapCompletionIdArg(id, 'resume')	2✔
529	const { table } = await this.getQueueCache(name)	2✔
530	const sql = plans.resumeJobs(this.config.schema, table)	2✔
531	const result = await db.executeSql(sql, [name, ids])	2✔
532	return this.mapCommandResponse(ids, result)	2✔
533	}
534
535	async retry (name, id, options = {}) {	1✔
536	Attorney.assertQueueName(name)	1✔
537	const db = options.db \|\| this.db	1✔
538	const ids = this.mapCompletionIdArg(id, 'retry')	1✔
539	const { table } = await this.getQueueCache(name)	1✔
540	const sql = plans.retryJobs(this.config.schema, table)	1✔
541	const result = await db.executeSql(sql, [name, ids])	1✔
542	return this.mapCommandResponse(ids, result)	1✔
543	}
544
545	async createQueue (name, options = {}) {	156✔
546	name = name \|\| options.name	178!
547
548	Attorney.assertQueueName(name)	178✔
549
550	options.policy = options.policy \|\| QUEUE_POLICIES.standard	178✔
551
552	assert(options.policy in QUEUE_POLICIES, `${options.policy} is not a valid queue policy`)	178✔
553
554	Attorney.validateQueueArgs(options)	177✔
555
556	if (options.deadLetter) {	177✔
557	Attorney.assertQueueName(options.deadLetter)	4✔
558	assert.notStrictEqual(name, options.deadLetter, 'deadLetter cannot be itself')	4✔
559	await this.getQueueCache(options.deadLetter)	4✔
560	}
561
562	const sql = plans.createQueue(this.config.schema, name, options)	177✔
563	await this.db.executeSql(sql)	177✔
564	}
565
566	async getQueues (names) {
567	if (names) {	210✔
568	names = Array.isArray(names) ? names : [names]	7!
569	for (const name of names) {	7✔
570	Attorney.assertQueueName(name)	7✔
571	}
572	}
573
574	const sql = plans.getQueues(this.config.schema, names)	210✔
575	const { rows } = await this.db.executeSql(sql)	210✔
576	return rows	210✔
577	}
578
579	async updateQueue (name, options = {}) {	×
580	Attorney.assertQueueName(name)	3✔
581
582	assert(Object.keys(options).length > 0, 'no properties found to update')	3✔
583
584	if ('policy' in options) {	3✔
585	assert(options.policy in QUEUE_POLICIES, `${options.policy} is not a valid queue policy`)	1✔
586	}
587
588	Attorney.validateQueueArgs(options)	3✔
589
590	const { deadLetter } = options	3✔
591
592	if (deadLetter) {	3!
593	Attorney.assertQueueName(deadLetter)	3✔
594	assert.notStrictEqual(name, deadLetter, 'deadLetter cannot be itself')	3✔
595	}
596
597	const sql = plans.updateQueue(this.config.schema, { deadLetter })	3✔
598	await this.db.executeSql(sql, [name, options])	3✔
599	}
600
601	async getQueue (name) {
602	Attorney.assertQueueName(name)	161✔
603
604	const sql = plans.getQueues(this.config.schema, [name])	161✔
605	const { rows } = await this.db.executeSql(sql)	161✔
606
607	return rows[0] \|\| null	161✔
608	}
609
610	async deleteQueue (name) {
611	Attorney.assertQueueName(name)	5✔
612
613	try {	5✔
614	await this.getQueueCache(name)	5✔
615	const sql = plans.deleteQueue(this.config.schema, name)	5✔
616	await this.db.executeSql(sql)	5✔
617	} catch {}
618	}
619
620	async deleteQueuedJobs (name) {
621	Attorney.assertQueueName(name)	1✔
622	const { table } = await this.getQueueCache(name)	1✔
623	const sql = plans.deleteQueuedJobs(this.config.schema, table)	1✔
624	await this.db.executeSql(sql, [name])	1✔
625	}
626
627	async deleteStoredJobs (name) {
628	Attorney.assertQueueName(name)	1✔
629	const { table } = await this.getQueueCache(name)	1✔
630	const sql = plans.deleteStoredJobs(this.config.schema, table)	1✔
631	await this.db.executeSql(sql, [name])	1✔
632	}
633
634	async deleteAllJobs (name) {
635	Attorney.assertQueueName(name)	3✔
636	const { table, partition } = await this.getQueueCache(name)	3✔
637
638	if (partition) {	3✔
639	const sql = plans.deleteAllJobs(this.config.schema, table)	2✔
640	await this.db.executeSql(sql, [name])	2✔
641	} else {
642	const sql = plans.truncateTable(this.config.schema, table)	1✔
643	await this.db.executeSql(sql)	1✔
644	}
645	}
646
647	async getQueueStats (name) {
648	Attorney.assertQueueName(name)	2✔
649
650	const { table } = await this.getQueueCache(name)	2✔
651
652	const sql = plans.getQueueStats(this.config.schema, table, [name])	2✔
653
654	const { rows } = await this.db.executeSql(sql)	2✔
655
656	return rows.at(0) \|\| null	2!
657	}
658
659	async getJobById (name, id, options = {}) {	39✔
660	Attorney.assertQueueName(name)	41✔
661
662	const db = this.assertDb(options)	41✔
663
664	const { table } = await this.getQueueCache(name)	41✔
665
666	const sql = plans.getJobById(this.config.schema, table)	41✔
667
668	const result1 = await db.executeSql(sql, [name, id])	41✔
669
670	if (result1?.rows?.length === 1) {	41✔
671	return result1.rows[0]	36✔
672	} else {
673	return null	5✔
674	}
675	}
676
677	assertDb (options) {
678	if (options.db) {	367✔
679	return options.db	10✔
680	}
681
682	assert(this.db._pgbdb && this.db.opened, 'Database connection is not opened')	357✔
683
684	return this.db	357✔
685	}
686	}
687
688	module.exports = Manager	4✔

timgit / pg-boss / 17884078814

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