17963854843

Committed 24 Sep 2025 01:46AM UTC coverage: 94.553% (-5.4%) from 100.0%

Build # 17963854843

Build Type

Pull #495

github

Committed by

web-flow

Commit Message

Merge de986c5f7 into 60d830619

Pull Request Pull Request #495: v11

Run Details

390 of 451 branches covered (86.47%)

320 of 321 new or added lines in 8 files covered. (99.69%)

49 existing lines in 4 files now uncovered.

868 of 918 relevant lines covered (94.55%)

107.87 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, policy, singletonsActive } = await this.getQueueCache(name)

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

timgit / pg-boss / 17963854843

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