Merge pull request #763 from bcomnes/expose-wip

Expose wip

Author: timgit

docs/api/workers.md

@@ -4,7 +4,7 @@
 
 Adds a new polling worker for a queue and executes the provided callback function when jobs are found. Each call to work() will add a new worker and resolve a unqiue worker id.
 
-Workers can be stopped via `offWork()` all at once by queue name or individually by using the worker id. Worker activity may be monitored by listening to the `wip` event.
+Workers can be stopped via `offWork()` all at once by queue name or individually by using the worker id. Worker activity may be monitored by listening to the `wip` event or by polling [`getWipData()`](#getwipdataoptions).
 
 The default options for `work()` is 1 job every 2 seconds.
 
@@ -217,6 +217,33 @@ await boss.work('process-video', async ([ job ]) => {
 })
 ```
 
+### `getWipData(options)`
+
+Returns a snapshot of all workers in this instance of pg-boss with state `created`, `active`, or `stopping`. This is the same data payload emitted by the `wip` event, but available on-demand without waiting for a job transition.
+
+Use this for continuous monitoring of worker utilization — for example, driving metrics or autoscaling signals when jobs are long-running and the `wip` event may not fire frequently enough.
+
+**Arguments**
+- `options`: object *(optional)*
+
+**Options**
+
+* **includeInternal**, bool, *(default=false)*
+
+  If true, includes workers for pg-boss internal queues (e.g., scheduling).
+
+**Returns**: `WipData[]`
+
+```js
+// Poll worker utilization every 2 seconds for metrics
+setInterval(() => {
+  const workers = boss.getWipData()
+  const working = workers.filter(w => w.state === 'active' && w.count > 0).length
+  const idle = workers.filter(w => w.state === 'active' && w.count === 0).length
+  console.log(`working: ${working}, idle: ${idle}`)
+}, 2000)
+```
+
 ### `notifyWorker(id)`
 
 Notifies a worker by id to bypass the job polling interval (see `pollingIntervalSeconds`) for this iteration in the loop.

src/index.ts

@@ -327,6 +327,10 @@ export class PgBoss extends EventEmitter<types.PgBossEventMap> {
     return this.#boss.supervise(name)
   }
 
+  getWipData (options?: { includeInternal?: boolean }): types.WipData[] {
+    return this.#manager.getWipData(options)
+  }
+
   getSpy<T = object> (name: string): JobSpyInterface<T> {
     return this.#manager.getSpy<T>(name)
   }

test/workTest.ts

@@ -328,6 +328,24 @@ describe('work', function () {
     expect(wip2.length).toBe(1)
   })
 
+  it('getWipData() should return current worker state', async function () {
+    ctx.boss = await helper.start(ctx.bossConfig)
+
+    await ctx.boss.send(ctx.schema)
+
+    await ctx.boss.work(ctx.schema, { pollingIntervalSeconds: 1 }, () => delay(2000))
+
+    // Wait for the job to be picked up
+    const firstWipEvent = new Promise<void>(resolve => ctx.boss!.once('wip', () => resolve()))
+    await firstWipEvent
+
+    const wip = ctx.boss.getWipData()
+
+    expect(wip.length).toBe(1)
+    expect(wip[0].name).toBe(ctx.schema)
+    expect(wip[0].state).toBe('active')
+  })
+
   it('should reject work() after stopping', async function () {
     ctx.boss = await helper.start(ctx.bossConfig)