Merge branch 'master' into corelate-wip-to-worker

Author: timgit

README.md

@@ -70,6 +70,48 @@ A HTTP proxy is available in the [`@pg-boss/proxy`](https://www.npmjs.com/packag
 
 See the [proxy documentation](https://github.com/timgit/pg-boss/blob/master/packages/proxy/README.md) for full configuration and deployment options.
 
+## ORM Transaction Adapters
+
+pg-boss ships adapters for running operations inside ORM-managed transactions.  Each adapter wraps the ORM's transaction object as an `IDatabase` you can pass via the `db` option on `send()`, `insert()`, `fetch()`, `complete()`, and other methods.
+
+### Knex / Kysely / Prisma
+
+```ts
+import { fromKnex, fromKysely, fromPrisma } from 'pg-boss'
+```
+
+```ts
+// Knex
+await knex.transaction(async (trx) => {
+  await boss.send('my-queue', data, { db: fromKnex(trx) })
+})
+
+// Kysely
+await db.transaction().execute(async (trx) => {
+  await boss.send('my-queue', data, { db: fromKysely(trx) })
+})
+
+// Prisma (v7+ with @prisma/adapter-pg)
+await prisma.$transaction(async (tx) => {
+  await boss.send('my-queue', data, { db: fromPrisma(tx) })
+})
+```
+
+### Drizzle
+
+The Drizzle adapter accepts the `sql` tagged-template function from `drizzle-orm` as a second argument so it can construct parameterised queries without a runtime dependency on `drizzle-orm`.
+
+```ts
+import { fromDrizzle } from 'pg-boss'
+import { sql } from 'drizzle-orm'
+```
+
+```ts
+await db.transaction(async (tx) => {
+  await boss.send('my-queue', data, { db: fromDrizzle(tx, sql) })
+})
+```
+
 ## Requirements
 * Node 22.12 or higher for CommonJS's require(esm)
 * PostgreSQL 13 or higher

docs/_sidebar.md

@@ -13,5 +13,6 @@
 * * [PubSub](./api/pubsub.md)
 * * [Workers](./api/workers.md)
 * * [Testing](./api/testing.md)
+* * [Adapters](./api/adapters.md)
 * * [Utils](./api/utils.md)
 * [SQL](sql.md)

docs/api/adapters.md

@@ -0,0 +1,70 @@
+# ORM Transaction Adapters
+
+pg-boss operations such as `send()`, `insert()`, `fetch()`, and `complete()` accept a `db` option that lets you run them inside an existing database transaction. This is how you ensure that job creation (or completion) is atomic with your application's own writes — if the transaction rolls back, so does the job.
+
+Each adapter wraps the ORM's transaction object as a pg-boss `Db` (the `executeSql` interface), so pg-boss can execute its own SQL within your transaction.
+
+```ts
+interface Db {
+  executeSql(text: string, values: any[]): Promise<{ rows: any[] }>;
+}
+```
+
+## Knex
+
+```ts
+import { fromKnex } from 'pg-boss'
+
+await knex.transaction(async (trx) => {
+  // your application writes ...
+  await trx('orders').insert({ item: 'widget', qty: 1 })
+
+  // schedule a pg-boss job in the same transaction
+  await boss.send('order-processing', { item: 'widget' }, { db: fromKnex(trx) })
+})
+```
+
+## Kysely
+
+```ts
+import { fromKysely } from 'pg-boss'
+
+await db.transaction().execute(async (trx) => {
+  await trx.insertInto('orders').values({ item: 'widget', qty: 1 }).execute()
+
+  await boss.send('order-processing', { item: 'widget' }, { db: fromKysely(trx) })
+})
+```
+
+## Drizzle
+
+The Drizzle adapter requires the `sql` tagged-template function from `drizzle-orm` as a second argument. This allows pg-boss to construct parameterised queries through Drizzle's public API without adding `drizzle-orm` as a runtime dependency.
+
+```ts
+import { fromDrizzle } from 'pg-boss'
+import { sql } from 'drizzle-orm'
+
+await db.transaction(async (tx) => {
+  await tx.insert(orders).values({ item: 'widget', qty: 1 })
+
+  await boss.send('order-processing', { item: 'widget' }, { db: fromDrizzle(tx, sql) })
+})
+```
+
+## Prisma
+
+Requires Prisma v7+ with `@prisma/adapter-pg`.
+
+```ts
+import { fromPrisma } from 'pg-boss'
+
+await prisma.$transaction(async (tx) => {
+  await tx.order.create({ data: { item: 'widget', qty: 1 } })
+
+  await boss.send('order-processing', { item: 'widget' }, { db: fromPrisma(tx) })
+})
+```
+
+## Rollback behaviour
+
+When the ORM transaction is rolled back (either explicitly or by throwing an error), all pg-boss operations executed through the adapter are rolled back as well. This is the primary reason to use these adapters — to guarantee atomicity between your application writes and job scheduling.

docs/api/jobs.md

@@ -80,6 +80,8 @@ All retry, expiration, and retention options can also be set on the queue and wi
   }
     ```
 
+  pg-boss ships with built-in adapters for popular ORMs. See [ORM Transaction Adapters](./api/adapters.md) for details.
+
 **Deferred jobs**
 
 * **startAfter** int, string, or Date

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.
 
@@ -13,7 +13,7 @@ The default options for `work()` is 1 job every 2 seconds.
 **Arguments**
 - `name`: string, *required*
 - `options`: object
-- `handler`: function(jobs), *required*
+- `handler`: function(jobs): Promise<any>, *required*
 
 **Options**
 
@@ -165,7 +165,7 @@ In this setup:
 
 **Handler function**
 
-`handler` should return a promise (Usually this is an `async` function). If an unhandled error occurs in a handler, `fail()` will automatically be called for the jobs, storing the error in the `output` property, making the job or jobs available for retry.
+`handler` should return a promise (Usually this is an `async` function). If the `handler` returns a value or an object, it will be stored in the `output` property. If an unhandled error occurs in a handler, `fail()` will automatically be called for the jobs, storing the error in the `output` property, making the job or jobs available for retry.
 
 The jobs argument is an array of jobs with the following properties.
 
@@ -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.