# Workers (/docs/code/jobs/workers)

---
title: Workers
description: Publishing a job only enqueues it. This is the entrypoint you write to drain the queue, in a long-running process, a serverless tick, or a pushed batch.
kind: guide
package: questpie
---

QUESTPIE does not start a worker for you. You write a small entrypoint that
imports the built app and calls one of three methods. Which one you can call
depends on the adapter, and calling the wrong one throws.

| Model                | Call                             | Works on          |
| -------------------- | -------------------------------- | ----------------- |
| Long-running process | `app.queue.listen()`             | pg-boss, BullMQ   |
| One batch, then exit | `app.queue.runOnce()`            | pg-boss, BullMQ   |
| Pushed batches       | `app.queue.createPushConsumer()` | Cloudflare Queues |

## A long-running worker

Write a file, run it as its own process.

```ts title="src/worker.ts"
import { app } from "#questpie";

await app.queue.listen({ teamSize: 5, batchSize: 3 });

console.log("Worker listening for jobs...");
```

Start it with `bun run src/worker.ts`.

`listen()` does five things before it returns.

- Registers every `options.cron` schedule, so recurring jobs exist from boot.
- Creates each pg-boss queue with its job's declared `queuePolicy`.
- Drains any dispatch intents left over from a crash.
- Starts consuming, and re-drains intents every five seconds.
- Installs `SIGINT` and `SIGTERM` handlers.

It resolves to a handle whose `stop()` removes the signal handlers and shuts the
adapter down.

The two options land differently per adapter.

| Option      | pg-boss                                 | BullMQ                            |
| ----------- | --------------------------------------- | --------------------------------- |
| `teamSize`  | Ignored. pg-boss 12 has no such option. | Becomes the worker `concurrency`. |
| `batchSize` | Jobs fetched per poll.                  | Ignored.                          |

QUESTPIE runs a fetched pg-boss batch one job at a time. To get more throughput
there, run more worker processes.

<Callout type="info" title="Shutdown is handled by default">
	Set `gracefulShutdown: false` to opt out. `shutdownSignals` changes which
	signals are caught. `shutdownTimeoutMs` defaults to 10000, after which the
	process is force-exited.
</Callout>

## One batch, then exit

A serverless function or a cron tick should not hold a listener open. Drain one
bounded batch and return.

```ts title="A scheduled serverless function"
import { app } from "#questpie";

export async function handler() {
	const { processed } = await app.queue.runOnce({ batchSize: 25 });
	return { processed };
}
```

`runOnce()` relays pending dispatch intents, processes one adapter batch, then
relays again. It resolves to `{ processed }`, the count. `batchSize` defaults to 10.

Pass `jobs: [...]` to restrict it to some of your jobs. Either the registration
key or the durable name works, and it normalises both.

## Pushed batches

Cloudflare Queues pushes to you. A Worker's `queue` export is the entrypoint,
and `questpie/adapters/cloudflare` builds it for you.

```ts title="Cloudflare Worker"
import { createCloudflareWorkerHandlers } from "questpie/adapters/cloudflare";

import { app } from "#questpie";

export default createCloudflareWorkerHandlers(app);
```

That gives you `fetch`, `queue` and `scheduled` at once. Take
`createCloudflareQueueHandler(app)` on its own if you already export the other
two. Both call `createPushConsumer()` and translate Cloudflare's batch into the
shape it wants, so you never cast.

Every delivery drains pending intents first, then hands the batch to the
adapter. Nothing runs when no batch arrives. So add a Cron Trigger calling
`await app.queue.drain()`, or committed work can sit forever. See
[Transactional dispatch](/docs/infrastructure/queue/transactional-dispatch).

## Check before you call

Each adapter advertises what it supports on `app.queue.capabilities`. Read the
flag instead of catching the throw when one codebase targets more than one
runtime.

```ts
if (app.queue.capabilities.longRunningConsumer) {
	await app.queue.listen();
}
```

| Flag                     | Gates                              |
| ------------------------ | ---------------------------------- |
| `longRunningConsumer`    | `listen()`                         |
| `runOnceConsumer`        | `runOnce()`                        |
| `pushConsumer`           | `createPushConsumer()`             |
| `scheduling`             | `schedule()`, `unschedule()`       |
| `singleton`              | Advertises `singletonKey` support. |
| `executionTerminalState` | `secretPayload`                    |

`scheduling` is the odd one. `.schedule()` throws when it is false, but
`options.cron` registration skips quietly instead. So on Cloudflare Queues a
cron job is never registered and nothing complains. The `scheduled` handler
above is how you get it back. It publishes every job whose `options.cron`
matches the trigger's own cron string, so declare the same expression in
`wrangler.toml`.

`questpie/queue` also exports `startJobWorker(app.queue, options)` and
`runJobWorkerOnce(app.queue, options)`. They wrap `listen()` and `runOnce()` and
add nothing. `startJobWorker` returns `void` rather than the handle, so call
`listen()` directly when you want `stop()`.

<Callout type="warn" title="Delivery is at least once">
	A crash between broker acceptance and the saved receipt can republish. Retries
	carry the same `dispatchId`, but the broker can still deliver twice. Make
	handlers idempotent.
</Callout>