# bunqueue Use Cases: Background Job Patterns for Bun

Six production patterns with copy-paste code: email delivery, webhooks, image processing, payments, cron scheduling, and multi-step job flows.

Canonical: https://bunqueue.dev/guide/use-cases/

---

import { Tabs, TabItem } from '@astrojs/starlight/components';

<div class="bq-wrap bq-hero">
  <span class="bq-eyebrow">guide · use-cases</span>
  <h1 class="bq-hero-h1 bq-bench-h1">The use cases teams <em>run.</em></h1>
  <p class="bq-hero-sub">Six patterns you can copy into a real app: emails, webhooks, images, payments, cron and multi-step flows. Each one moves slow work out of your API request and into a background job.</p>
</div>

This page shows the most common things people build with bunqueue, each with a small working example. If a term is new to you, it gets a plain-words explanation the first time it appears.

## The core pattern

Every use case below is a variation of the same idea: your API handler adds a job (a unit of work saved as data) to a queue and returns immediately. A worker (a function that pulls jobs and runs them) does the slow part in the background.

<Tabs syncKey="lang">
<TabItem label="Bun">

```typescript
import { Queue, Worker } from 'bunqueue/client';

// Embedded mode: the queue runs inside your process, no separate server.
// dataPath persists jobs to a SQLite file so they survive restarts.
const queue = new Queue('emails', { embedded: true, dataPath: './data/app.db' });

new Worker('emails', async (job) => {
  await sendEmail(job.data);
  return { sent: true };
}, { embedded: true, concurrency: 10 }); // up to 10 jobs in parallel

// In your API handler: this returns in microseconds
await queue.add('welcome', { to: 'user@example.com' });
```

</TabItem>
<TabItem label="Node.js / Deno">

```typescript
import { Queue, Worker } from 'bunqueue-client';

// TCP mode: the broker owns persistence; configure its database on the server.
const queue = new Queue('emails', { embedded: false });

new Worker('emails', async (job) => {
  await sendEmail(job.data);
  return { sent: true };
}, { embedded: false, concurrency: 10 }); // up to 10 jobs in parallel

// In your API handler: this resolves after the broker accepts the job
await queue.add('welcome', { to: 'user@example.com' });
```

</TabItem>
<TabItem label="Python">

```python
from bunqueue import Queue, Worker

# Connects to a bunqueue server on localhost:6789
queue = Queue("emails")

def process(job):
    send_email(job.data)
    return {"sent": True}

Worker("emails", process, concurrency=10)  # up to 10 jobs in parallel

# In your API handler: this returns as soon as the job is queued
queue.add("welcome", {"to": "user@example.com"})
```

</TabItem>
<TabItem label="PHP">

```php
use Bunqueue\Queue;
use Bunqueue\Worker;

// Connects to a bunqueue server on localhost:6789
$queue = new Queue('emails');

// In your API handler: this returns as soon as the job is queued
$queue->add('welcome', ['to' => 'user@example.com']);

// worker.php, a separate long-running process
$worker = new Worker('emails', function (Bunqueue\Job $job) {
    sendEmail($job->data());
    return ['sent' => true];
});
$worker->run();
```

</TabItem>
<TabItem label="Go">

```go
// Connects to a bunqueue server on localhost:6789
queue := bunqueue.NewQueue("emails", bunqueue.Options{})
defer queue.Close()

// In your API handler: this returns as soon as the job is queued
queue.Add("welcome", map[string]any{"to": "user@example.com"}, nil)

// worker, a separate long-running process
worker := bunqueue.NewWorker("emails", func(job *bunqueue.Job) (any, error) {
    return sendEmail(job.Data())
}, bunqueue.WorkerOptions{Concurrency: 10}) // up to 10 jobs in parallel
worker.Run()
```

</TabItem>
<TabItem label="Rust">

```rust
use bunqueue_client::{ConnectionOptions, JobOptions, ProcessError, Queue, Value, Worker, WorkerOptions};

// Connects to a bunqueue server on localhost:6789
let queue = Queue::new("emails", ConnectionOptions::default());

// In your API handler: this returns as soon as the job is queued
let data = Value::Map(vec![(Value::from("to"), Value::from("user@example.com"))]);
queue.add("welcome", data, JobOptions::default())?;

// worker, a separate long-running process
let worker = Worker::new(
    "emails",
    |job| {
        deliver(job.data())
            .map(|_| Value::from(true))
            .map_err(|error| ProcessError::retryable(error.to_string()))
    },
    WorkerOptions { concurrency: 10, ..Default::default() }, // up to 10 jobs in parallel
);
worker.run()?;
```

</TabItem>
<TabItem label="Elixir">

```elixir
# Connects to a bunqueue server on localhost:6789
queue = Bunqueue.queue("emails")

# In your API handler: this returns as soon as the job is queued
{:ok, _job} = Bunqueue.Queue.add(queue, "welcome", %{to: "user@example.com"})

# worker, a separate long-running process
worker =
  Bunqueue.Worker.new("emails", fn job ->
    send_email(job.data)
    {:ok, %{sent: true}}
  end, concurrency: 10)

Bunqueue.Worker.run(worker)
```

</TabItem>
</Tabs>

That is the whole model. The sections below add the options that make each use case reliable. New to bunqueue? Start with the [quickstart](/guide/quickstart/).

## Email delivery

Sending email inside an API request is slow and fragile: the provider can be down or rate limited. Queue it instead, and let bunqueue retry on failure.

<Tabs syncKey="lang">
<TabItem label="Bun">

```typescript
import { Queue, Worker } from 'bunqueue/client';

interface EmailJob { to: string; template: string; data: Record<string, unknown> }

const emails = new Queue<EmailJob>('emails', {
  embedded: true,
  dataPath: './data/app.db',
  defaultJobOptions: {
    attempts: 5,           // try up to 5 times
    backoff: 2000,         // wait 2s, then 4s, 8s... between tries (exponential backoff)
    removeOnComplete: true, // drop finished jobs to keep the queue lean
  },
});

new Worker<EmailJob>('emails', async (job) => {
  const result = await sendEmail(job.data);
  return { messageId: result.messageId };
}, { embedded: true, concurrency: 10 });

// One email
await emails.add('welcome', { to: 'user@example.com', template: 'welcome', data: { name: 'John' } });

// Bulk newsletter, batched in one call
await emails.addBulk(
  subscribers.map((s) => ({ name: 'newsletter', data: { to: s.email, template: 'news', data: {} } }))
);
```

</TabItem>
<TabItem label="Node.js / Deno">

```typescript
import { Queue, Worker } from 'bunqueue-client';

interface EmailJob { to: string; template: string; data: Record<string, unknown> }

const emails = new Queue<EmailJob>('emails', {
  embedded: false,
  defaultJobOptions: {
    attempts: 5,           // try up to 5 times
    backoff: 2000,         // wait 2s, then 4s, 8s... between tries (exponential backoff)
    removeOnComplete: true, // drop finished jobs to keep the queue lean
  },
});

new Worker<EmailJob>('emails', async (job) => {
  const result = await sendEmail(job.data);
  return { messageId: result.messageId };
}, { embedded: false, concurrency: 10 });

// One email
await emails.add('welcome', { to: 'user@example.com', template: 'welcome', data: { name: 'John' } });

// Bulk newsletter, batched in one call
await emails.addBulk(
  subscribers.map((s) => ({ name: 'newsletter', data: { to: s.email, template: 'news', data: {} } }))
);
```

</TabItem>
<TabItem label="Python">

```python
from bunqueue import Queue, Worker

emails = Queue("emails")

# Retry options are passed per add
retry_opts = {
    "attempts": 5,              # try up to 5 times
    "backoff": 2000,            # wait 2s, then 4s, 8s... between tries (exponential backoff)
    "remove_on_complete": True, # drop finished jobs to keep the queue lean
}

def process(job):
    result = send_email(job.data)
    return {"message_id": result["message_id"]}

Worker("emails", process, concurrency=10)

# One email
emails.add("welcome",
           {"to": "user@example.com", "template": "welcome", "data": {"name": "John"}},
           **retry_opts)

# Bulk newsletter, batched in one call
emails.add_bulk([
    {"name": "newsletter", "data": {"to": s["email"], "template": "news", "data": {}}, **retry_opts}
    for s in subscribers
])
```

</TabItem>
<TabItem label="PHP">

Not shown here. The PHP equivalent is in the [PHP SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/php) and the [SDK guide](/guide/sdks/).

</TabItem>
<TabItem label="Go">

```go
emails := bunqueue.NewQueue("emails", bunqueue.Options{})
defer emails.Close()

// Retry options are passed per add
retryOpts := bunqueue.JobOptions{
    "attempts":         5,    // try up to 5 times
    "backoff":          2000, // wait 2s, then 4s, 8s... between tries (exponential backoff)
    "removeOnComplete": true, // drop finished jobs to keep the queue lean
}

worker := bunqueue.NewWorker("emails", func(job *bunqueue.Job) (any, error) {
    result, err := sendEmail(job.Data())
    if err != nil {
        return nil, err // retried automatically
    }
    return map[string]any{"messageId": result.MessageID}, nil
}, bunqueue.WorkerOptions{Concurrency: 10})
go worker.Run()

// One email
emails.Add("welcome",
    map[string]any{"to": "user@example.com", "template": "welcome", "data": map[string]any{"name": "John"}},
    retryOpts)

// Bulk newsletter, batched in one call
entries := make([]bunqueue.BulkEntry, 0, len(subscribers))
for _, s := range subscribers {
    entries = append(entries, bunqueue.BulkEntry{
        Name: "newsletter",
        Data: map[string]any{"to": s.Email, "template": "news"},
        Opts: retryOpts,
    })
}
emails.AddBulk(entries)
```

</TabItem>
<TabItem label="Rust">

Not shown here. The Rust equivalent is in the [Rust SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/rust) and the [SDK guide](/guide/sdks/).

</TabItem>
<TabItem label="Elixir">

Not shown here. The Elixir equivalent is in the [Elixir SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/elixir) and the [SDK guide](/guide/sdks/).

</TabItem>
</Tabs>

*PHP, Rust and Elixir follow the same shape, `add` / `addBulk` on the producer and a worker with retries handled server-side, see the [SDK guide](/guide/sdks/).*

If all 5 attempts fail, the job lands in the dead letter queue (DLQ), a holding area for jobs that ran out of retries. Inspect it with `queue.getDlq()` and retry with `queue.retryDlq()`. See the [DLQ guide](/guide/dlq/).

## Webhook delivery

Partner endpoints go down and return 5xx errors. Treat every delivery as a job with retries, and let the DLQ auto-retry the stubborn ones on a schedule.

<Tabs syncKey="lang">
<TabItem label="Bun">

```typescript
const webhooks = new Queue('webhooks', {
  embedded: true,
  dataPath: './data/app.db',
  defaultJobOptions: { attempts: 8, backoff: 5000 }, // 5s, 10s, 20s, 40s...
});

// Jobs that exhaust all 8 attempts go to the DLQ.
// Auto-retry the DLQ every hour, up to 3 times, then keep entries 7 days.
webhooks.setDlqConfig({
  autoRetry: true,
  autoRetryInterval: 3_600_000,
  maxAutoRetries: 3,
  maxAge: 604_800_000,
});

new Worker('webhooks', async (job) => {
  const { endpoint, event, payload } = job.data;
  const res = await fetch(endpoint, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', 'X-Webhook-Event': event },
    body: JSON.stringify(payload),
    signal: AbortSignal.timeout(30_000), // never hang on a dead endpoint
  });
  if (!res.ok) throw new Error(`HTTP ${res.status}`); // throwing triggers a retry
  return { status: res.status };
}, { embedded: true, concurrency: 20 });

await webhooks.add('order.created', {
  endpoint: 'https://partner.com/webhooks',
  event: 'order.created',
  payload: { orderId: 'ORD-123' },
});
```

</TabItem>
<TabItem label="Node.js / Deno">

```typescript
const webhooks = new Queue('webhooks', {
  embedded: false,
  defaultJobOptions: { attempts: 8, backoff: 5000 }, // 5s, 10s, 20s, 40s...
});

// Jobs that exhaust all 8 attempts go to the DLQ.
// Auto-retry the DLQ every hour, up to 3 times, then keep entries 7 days.
await webhooks.setDlqConfigAsync({
  autoRetry: true,
  autoRetryInterval: 3_600_000,
  maxAutoRetries: 3,
  maxAge: 604_800_000,
});

new Worker('webhooks', async (job) => {
  const { endpoint, event, payload } = job.data;
  const res = await fetch(endpoint, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', 'X-Webhook-Event': event },
    body: JSON.stringify(payload),
    signal: AbortSignal.timeout(30_000), // never hang on a dead endpoint
  });
  if (!res.ok) throw new Error(`HTTP ${res.status}`); // throwing triggers a retry
  return { status: res.status };
}, { embedded: false, concurrency: 20 });

await webhooks.add('order.created', {
  endpoint: 'https://partner.com/webhooks',
  event: 'order.created',
  payload: { orderId: 'ORD-123' },
});
```

</TabItem>
<TabItem label="Python">

```python
import requests
from bunqueue import Queue, Worker

webhooks = Queue("webhooks")

# Jobs that exhaust all 8 attempts go to the DLQ.
# Auto-retry the DLQ every hour, up to 3 times, then keep entries 7 days.
webhooks.set_dlq_config({
    "autoRetry": True,
    "autoRetryInterval": 3_600_000,
    "maxAutoRetries": 3,
    "maxAge": 604_800_000,
})

def deliver(job):
    res = requests.post(
        job.data["endpoint"],
        json=job.data["payload"],
        headers={"X-Webhook-Event": job.data["event"]},
        timeout=30,  # never hang on a dead endpoint
    )
    res.raise_for_status()  # raising triggers a retry
    return {"status": res.status_code}

Worker("webhooks", deliver, concurrency=20)

webhooks.add("order.created", {
    "endpoint": "https://partner.com/webhooks",
    "event": "order.created",
    "payload": {"orderId": "ORD-123"},
}, attempts=8, backoff=5000)  # 5s, 10s, 20s, 40s...
```

</TabItem>
<TabItem label="PHP">

Not shown here. The PHP equivalent is in the [PHP SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/php) and the [SDK guide](/guide/sdks/).

</TabItem>
<TabItem label="Go">

```go
webhooks := bunqueue.NewQueue("webhooks", bunqueue.Options{})
defer webhooks.Close()

// DLQ auto-retry configuration (setDlqConfig) is available from the
// Bun, TypeScript and Python clients.

worker := bunqueue.NewWorker("webhooks", func(job *bunqueue.Job) (any, error) {
    data := job.Data()
    payload, _ := json.Marshal(data["payload"])
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel() // never hang on a dead endpoint

    req, _ := http.NewRequestWithContext(ctx, "POST", data["endpoint"].(string), bytes.NewReader(payload))
    req.Header.Set("Content-Type", "application/json")
    req.Header.Set("X-Webhook-Event", data["event"].(string))
    res, err := http.DefaultClient.Do(req)
    if err != nil {
        return nil, err // returning an error triggers a retry
    }
    defer res.Body.Close()
    if res.StatusCode >= 400 {
        return nil, fmt.Errorf("HTTP %d", res.StatusCode)
    }
    return map[string]any{"status": res.StatusCode}, nil
}, bunqueue.WorkerOptions{Concurrency: 20})
go worker.Run()

webhooks.Add("order.created", map[string]any{
    "endpoint": "https://partner.com/webhooks",
    "event":    "order.created",
    "payload":  map[string]any{"orderId": "ORD-123"},
}, bunqueue.JobOptions{"attempts": 8, "backoff": 5000}) // 5s, 10s, 20s, 40s...
```

</TabItem>
<TabItem label="Rust">

Not shown here. The Rust equivalent is in the [Rust SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/rust) and the [SDK guide](/guide/sdks/).

</TabItem>
<TabItem label="Elixir">

Not shown here. The Elixir equivalent is in the [Elixir SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/elixir) and the [SDK guide](/guide/sdks/).

</TabItem>
</Tabs>

*PHP, Rust and Elixir deliver webhooks with the same worker shape (see the [SDK guide](/guide/sdks/)). DLQ auto-retry configuration (`setDlqConfig`) is available from the Bun, TypeScript and Python clients.*

## Image processing

Generating thumbnails and variants during an upload request makes the upload slow. Queue one job per image, report progress as each variant finishes, and cap the runtime with a timeout.

<Tabs syncKey="lang">
<TabItem label="Bun">

```typescript
const images = new Queue('images', {
  embedded: true,
  dataPath: './data/app.db',
  defaultJobOptions: { attempts: 3, timeout: 120_000 }, // kill stuck jobs after 2 minutes
});

new Worker('images', async (job) => {
  const { sourceUrl, variants } = job.data;
  const source = await downloadImage(sourceUrl);
  const urls: Record<string, string> = {};

  for (let i = 0; i < variants.length; i++) {
    const v = variants[i];
    // updateProgress(percent, message): your frontend can poll or subscribe to this
    await job.updateProgress(Math.round((i / variants.length) * 100), `Processing ${v.name}`);
    const out = await sharp(source).resize(v.width, v.height).webp().toBuffer();
    urls[v.name] = await uploadToCDN(out, `${job.id}/${v.name}.webp`);
  }

  await job.updateProgress(100, 'Done');
  return { urls };
}, { embedded: true, concurrency: 5 });

await images.add('product-image', {
  sourceUrl: 'https://uploads.example.com/raw/product-123.jpg',
  variants: [
    { name: 'thumb', width: 150, height: 150 },
    { name: 'full', width: 1200, height: 900 },
  ],
});
```

</TabItem>
<TabItem label="Node.js / Deno">

```typescript
const images = new Queue('images', {
  embedded: false,
  defaultJobOptions: { attempts: 3, timeout: 120_000 }, // kill stuck jobs after 2 minutes
});

new Worker('images', async (job) => {
  const { sourceUrl, variants } = job.data;
  const source = await downloadImage(sourceUrl);
  const urls: Record<string, string> = {};

  for (let i = 0; i < variants.length; i++) {
    const v = variants[i];
    // updateProgress(percent, message): your frontend can poll or subscribe to this
    await job.updateProgress(Math.round((i / variants.length) * 100), `Processing ${v.name}`);
    const out = await sharp(source).resize(v.width, v.height).webp().toBuffer();
    urls[v.name] = await uploadToCDN(out, `${job.id}/${v.name}.webp`);
  }

  await job.updateProgress(100, 'Done');
  return { urls };
}, { embedded: false, concurrency: 5 });

await images.add('product-image', {
  sourceUrl: 'https://uploads.example.com/raw/product-123.jpg',
  variants: [
    { name: 'thumb', width: 150, height: 150 },
    { name: 'full', width: 1200, height: 900 },
  ],
});
```

</TabItem>
<TabItem label="Python">

```python
from bunqueue import Queue, Worker

images = Queue("images")

def process(job):
    variants = job.data["variants"]
    source = download_image(job.data["sourceUrl"])
    urls = {}

    for i, v in enumerate(variants):
        # update_progress(percent, message): your frontend can poll this
        job.update_progress(round(i / len(variants) * 100), f"Processing {v['name']}")
        out = resize_image(source, v["width"], v["height"])  # e.g. Pillow
        urls[v["name"]] = upload_to_cdn(out, f"{job.id}/{v['name']}.webp")

    job.update_progress(100, "Done")
    return {"urls": urls}

Worker("images", process, concurrency=5)

images.add("product-image", {
    "sourceUrl": "https://uploads.example.com/raw/product-123.jpg",
    "variants": [
        {"name": "thumb", "width": 150, "height": 150},
        {"name": "full", "width": 1200, "height": 900},
    ],
}, attempts=3, timeout=120_000)  # kill stuck jobs after 2 minutes
```

</TabItem>
<TabItem label="PHP">

Not shown here. The PHP equivalent is in the [PHP SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/php) and the [SDK guide](/guide/sdks/).

</TabItem>
<TabItem label="Go">

```go
images := bunqueue.NewQueue("images", bunqueue.Options{})
defer images.Close()

worker := bunqueue.NewWorker("images", func(job *bunqueue.Job) (any, error) {
    data := job.Data()
    variants := data["variants"].([]any)
    source, err := downloadImage(data["sourceUrl"].(string))
    if err != nil {
        return nil, err
    }
    urls := map[string]string{}

    for i, raw := range variants {
        v := raw.(map[string]any)
        name := v["name"].(string)
        // UpdateProgress(percent, message): your frontend can poll this
        job.UpdateProgress(float64(i)/float64(len(variants))*100, "Processing "+name)
        out := resizeVariant(source, v) // your image library of choice
        urls[name] = uploadToCDN(out, job.ID()+"/"+name+".webp")
    }

    job.UpdateProgress(100, "Done")
    return map[string]any{"urls": urls}, nil
}, bunqueue.WorkerOptions{Concurrency: 5})
go worker.Run()

images.Add("product-image", map[string]any{
    "sourceUrl": "https://uploads.example.com/raw/product-123.jpg",
    "variants": []any{
        map[string]any{"name": "thumb", "width": 150, "height": 150},
        map[string]any{"name": "full", "width": 1200, "height": 900},
    },
}, bunqueue.JobOptions{"attempts": 3, "timeout": 120000}) // kill stuck jobs after 2 minutes
```

</TabItem>
<TabItem label="Rust">

Not shown here. The Rust equivalent is in the [Rust SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/rust) and the [SDK guide](/guide/sdks/).

</TabItem>
<TabItem label="Elixir">

Not shown here. The Elixir equivalent is in the [Elixir SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/elixir) and the [SDK guide](/guide/sdks/).

</TabItem>
</Tabs>

*PHP, Rust and Elixir report progress the same way (`updateProgress` / `update_progress`), see the [Worker guide](/guide/worker/job-object/#use-the-job-object).*

The same shape works for video transcoding or any long CPU-bound task, just raise the `timeout`. For heavy CPU work see [CPU-intensive workers](/guide/cpu-intensive-workers/).

## Payments and other critical jobs

A payment job must never be lost, and its external side effect must be
idempotent because bunqueue provides at-least-once delivery. Two job options
support that design:

- On SQLite, `durable: true` skips the 10ms write buffer and writes the job
  before `add()` returns. PostgreSQL admissions are already transactional and
  do not use that buffer.
- `jobId` makes enqueueing idempotent: adding the same unfinished `jobId` twice
  returns the existing job instead of creating another generation. Use the
  same key with the payment provider because a lease recovery can redeliver a
  job after its side effect completed but before its ACK committed.

<Tabs syncKey="lang">
<TabItem label="Bun">

```typescript
const payments = new Queue('payments', {
  embedded: true,
  dataPath: './data/payments.db', // durable writes need a SQLite file
  defaultJobOptions: { attempts: 3, backoff: 5000, timeout: 60_000 },
});

// Failed payments need a human, never an automatic re-charge
payments.setDlqConfig({ autoRetry: false, maxAge: 2_592_000_000 }); // keep 30 days

new Worker('payments', async (job) => {
  const { orderId, amount, idempotencyKey } = job.data;
  const intent = await stripe.paymentIntents.create(
    { amount, currency: 'usd', confirm: true },
    { idempotencyKey } // provider-side guard against double charges
  );
  if (intent.status !== 'succeeded') throw new Error(`Payment failed: ${intent.status}`);
  await recordTransaction(orderId, intent.id);
  return { paymentIntentId: intent.id };
}, { embedded: true, concurrency: 5 });

await payments.add(
  'charge',
  { orderId: 'ORD-123', amount: 9999, idempotencyKey: 'order-ORD-123' },
  // Custom ID deduplicates while retained; durable closes SQLite's admission buffer window.
  { jobId: 'charge-ORD-123', durable: true }
);
```

</TabItem>
<TabItem label="Node.js / Deno">

```typescript
const payments = new Queue('payments', {
  embedded: false,
  defaultJobOptions: { attempts: 3, backoff: 5000, timeout: 60_000 },
});

// Failed payments need a human, never an automatic re-charge
await payments.setDlqConfigAsync({ autoRetry: false, maxAge: 2_592_000_000 }); // keep 30 days

new Worker('payments', async (job) => {
  const { orderId, amount, idempotencyKey } = job.data;
  const intent = await stripe.paymentIntents.create(
    { amount, currency: 'usd', confirm: true },
    { idempotencyKey } // provider-side guard against double charges
  );
  if (intent.status !== 'succeeded') throw new Error(`Payment failed: ${intent.status}`);
  await recordTransaction(orderId, intent.id);
  return { paymentIntentId: intent.id };
}, { embedded: false, concurrency: 5 });

await payments.add(
  'charge',
  { orderId: 'ORD-123', amount: 9999, idempotencyKey: 'order-ORD-123' },
  // Custom ID deduplicates while retained; durable closes SQLite's admission buffer window.
  { jobId: 'charge-ORD-123', durable: true }
);
```

</TabItem>
<TabItem label="Python">

```python
from bunqueue import Queue, Worker

payments = Queue("payments")

# Failed payments need a human, never an automatic re-charge
payments.set_dlq_config({"autoRetry": False, "maxAge": 2_592_000_000})  # keep 30 days

def charge(job):
    intent = stripe.PaymentIntent.create(
        amount=job.data["amount"],
        currency="usd",
        confirm=True,
        idempotency_key=job.data["idempotencyKey"],  # guard against double charges
    )
    if intent.status != "succeeded":
        raise Exception(f"Payment failed: {intent.status}")
    record_transaction(job.data["orderId"], intent.id)
    return {"payment_intent_id": intent.id}

Worker("payments", charge, concurrency=5)

payments.add(
    "charge",
    {"orderId": "ORD-123", "amount": 9999, "idempotencyKey": "order-ORD-123"},
    job_id="charge-ORD-123", durable=True,  # SQLite closes its buffer window; PG is transactional
    attempts=3, backoff=5000, timeout=60_000,
)
```

</TabItem>
<TabItem label="PHP">

Not shown here. The PHP equivalent is in the [PHP SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/php) and the [SDK guide](/guide/sdks/).

</TabItem>
<TabItem label="Go">

```go
payments := bunqueue.NewQueue("payments", bunqueue.Options{})
defer payments.Close()

worker := bunqueue.NewWorker("payments", func(job *bunqueue.Job) (any, error) {
    data := job.Data()
    // Pass data["idempotencyKey"] to your provider as its idempotency key:
    // a provider-side guard against double charges.
    intent, err := chargeProvider(data)
    if err != nil {
        return nil, err
    }
    if err := recordTransaction(data["orderId"].(string), intent.ID); err != nil {
        return nil, err
    }
    return map[string]any{"paymentIntentId": intent.ID}, nil
}, bunqueue.WorkerOptions{Concurrency: 5})
go worker.Run()

payments.Add("charge",
    map[string]any{"orderId": "ORD-123", "amount": 9999, "idempotencyKey": "order-ORD-123"},
    bunqueue.JobOptions{
        "jobId":   "charge-ORD-123", // no duplicate
        "durable": true,             // closes SQLite's buffered-admission window; PG is already transactional
        "attempts": 3, "backoff": 5000, "timeout": 60000,
    })
```

</TabItem>
<TabItem label="Rust">

Not shown here. The Rust equivalent is in the [Rust SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/rust) and the [SDK guide](/guide/sdks/).

</TabItem>
<TabItem label="Elixir">

Not shown here. The Elixir equivalent is in the [Elixir SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/elixir) and the [SDK guide](/guide/sdks/).

</TabItem>
</Tabs>

*PHP, Rust and Elixir support `jobId` and `durable` identically, see the [SDK guide](/guide/sdks/#job-options). DLQ auto-retry configuration (`setDlqConfig`) is available from the Bun, TypeScript and Python clients.*

## Scheduled tasks (cron)

Recurring jobs use `upsertJobScheduler()`. Schedules survive restarts when the
selected backend is persistent: SQLite stores them in embedded or single-broker
server mode, while PostgreSQL stores and coordinates them across the broker
fleet. A cron expression like `0 3 * * *` means "every day at 3 AM".

<Tabs syncKey="lang">
<TabItem label="Bun">

```typescript
const scheduled = new Queue('scheduled', { embedded: true, dataPath: './data/app.db' });

// Daily cleanup at 3 AM
await scheduled.upsertJobScheduler(
  'daily-cleanup',
  { pattern: '0 3 * * *' },
  { data: { task: 'cleanup' } }
);

// Health check every 5 minutes
await scheduled.upsertJobScheduler(
  'health-check',
  { every: 300_000 },
  { data: { task: 'health-check' } }
);

new Worker('scheduled', async (job) => {
  if (job.data.task === 'cleanup') return { deleted: await cleanupOldRecords() };
  return await checkSystemHealth();
}, { embedded: true });
```

</TabItem>
<TabItem label="Node.js / Deno">

```typescript
const scheduled = new Queue('scheduled', { embedded: false });

// Daily cleanup at 3 AM
await scheduled.upsertJobScheduler(
  'daily-cleanup',
  { pattern: '0 3 * * *' },
  { data: { task: 'cleanup' } }
);

// Health check every 5 minutes
await scheduled.upsertJobScheduler(
  'health-check',
  { every: 300_000 },
  { data: { task: 'health-check' } }
);

new Worker('scheduled', async (job) => {
  if (job.data.task === 'cleanup') return { deleted: await cleanupOldRecords() };
  return await checkSystemHealth();
}, { embedded: false });
```

</TabItem>
<TabItem label="Python">

```python
from bunqueue import Queue, Worker

scheduled = Queue("scheduled")

# Daily cleanup at 3 AM
scheduled.upsert_job_scheduler("daily-cleanup", {"pattern": "0 3 * * *"},
                               {"data": {"task": "cleanup"}})

# Health check every 5 minutes
scheduled.upsert_job_scheduler("health-check", {"every": 300_000},
                               {"data": {"task": "health-check"}})

def process(job):
    if job.data["task"] == "cleanup":
        return {"deleted": cleanup_old_records()}
    return check_system_health()

Worker("scheduled", process)
```

</TabItem>
<TabItem label="PHP">

```php
$scheduled = new Bunqueue\Queue('scheduled');

// Daily cleanup at 3 AM
$scheduled->upsertJobScheduler('daily-cleanup',
    ['pattern' => '0 3 * * *'],
    ['data' => ['task' => 'cleanup']],
);

// Health check every 5 minutes
$scheduled->upsertJobScheduler('health-check',
    ['every' => 300000],
    ['data' => ['task' => 'health-check']],
);

$worker = new Bunqueue\Worker('scheduled', function (Bunqueue\Job $job) {
    if ($job->data()['task'] === 'cleanup') {
        return ['deleted' => cleanupOldRecords()];
    }
    return checkSystemHealth();
});
$worker->run();
```

</TabItem>
<TabItem label="Go">

```go
scheduled := bunqueue.NewQueue("scheduled", bunqueue.Options{})
defer scheduled.Close()

// Daily cleanup at 3 AM
scheduled.UpsertJobScheduler("daily-cleanup",
    bunqueue.SchedulerRepeat{Pattern: "0 3 * * *"},
    bunqueue.SchedulerTemplate{Data: map[string]any{"task": "cleanup"}},
)

// Health check every 5 minutes
scheduled.UpsertJobScheduler("health-check",
    bunqueue.SchedulerRepeat{EveryMs: 300000},
    bunqueue.SchedulerTemplate{Data: map[string]any{"task": "health-check"}},
)

worker := bunqueue.NewWorker("scheduled", func(job *bunqueue.Job) (any, error) {
    if job.Data()["task"] == "cleanup" {
        return map[string]any{"deleted": cleanupOldRecords()}, nil
    }
    return checkSystemHealth()
}, bunqueue.WorkerOptions{})
worker.Run()
```

</TabItem>
<TabItem label="Rust">

```rust
use bunqueue_client::{ConnectionOptions, Queue, SchedulerRepeat, SchedulerTemplate, Value};

let scheduled = Queue::new("scheduled", ConnectionOptions::default());

// Daily cleanup at 3 AM
scheduled.upsert_job_scheduler(
    "daily-cleanup",
    SchedulerRepeat { pattern: Some("0 3 * * *".into()), ..Default::default() },
    SchedulerTemplate {
        data: Value::Map(vec![(Value::from("task"), Value::from("cleanup"))]),
        ..Default::default()
    },
)?;

// Health check every 5 minutes
scheduled.upsert_job_scheduler(
    "health-check",
    SchedulerRepeat { every_ms: Some(300_000), ..Default::default() },
    SchedulerTemplate {
        data: Value::Map(vec![(Value::from("task"), Value::from("health-check"))]),
        ..Default::default()
    },
)?;
```

</TabItem>
<TabItem label="Elixir">

```elixir
scheduled = Bunqueue.queue("scheduled")

# Daily cleanup at 3 AM
:ok =
  Bunqueue.Queue.upsert_scheduler(scheduled, "daily-cleanup",
    %{pattern: "0 3 * * *"},
    %{data: %{task: "cleanup"}}
  )

# Health check every 5 minutes
:ok =
  Bunqueue.Queue.upsert_scheduler(scheduled, "health-check",
    %{every: 300_000},
    %{data: %{task: "health-check"}}
  )

worker =
  Bunqueue.Worker.new("scheduled", fn job ->
    case job.data["task"] do
      "cleanup" -> {:ok, %{deleted: cleanup_old_records()}}
      _ -> {:ok, check_system_health()}
    end
  end)

Bunqueue.Worker.run(worker)
```

</TabItem>
</Tabs>

Timezones, one-off delayed jobs and the `repeat` shorthand on `queue.add()` are covered in the [cron guide](/guide/cron/).

## Multi-step flows

Some work has dependencies: an order ships only after inventory and payment both check out. `FlowProducer` runs child jobs first, in parallel, then runs the parent with access to every child result.

<Tabs syncKey="lang">
<TabItem label="Bun">

```typescript
import { FlowProducer, Worker } from 'bunqueue/client';

type OrderData = { orderId: string };
const flow = new FlowProducer({ embedded: true });

const checks = new Worker<OrderData>('checks', async (job) => ({
  check: job.name,
  approved: true,
}), { embedded: true });

const orders = new Worker<OrderData>('orders', async (job) => {
  // Children finished first; read what each one returned
  const results = await job.getChildrenValues();
  return { orderId: job.data.orderId, shipped: true, checks: results };
}, { embedded: true });

const node = await flow.add<OrderData>({
  name: 'fulfill-order',
  queueName: 'orders',
  data: { orderId: 'ORD-123' },
  children: [
    { name: 'check-inventory', queueName: 'checks', data: { orderId: 'ORD-123' } },
    { name: 'check-payment', queueName: 'checks', data: { orderId: 'ORD-123' } },
  ],
});

const result = await node.job.waitUntilFinished(null, 10_000);
console.log(result);

await checks.close();
await orders.close();
await flow.close();
```

</TabItem>
<TabItem label="Node.js / Deno">

```typescript
import { FlowProducer, Worker } from 'bunqueue-client';

type OrderData = { orderId: string };
const flow = new FlowProducer({ embedded: false });

const checks = new Worker<OrderData>('checks', async (job) => ({
  check: job.name,
  approved: true,
}), { embedded: false });

const orders = new Worker<OrderData>('orders', async (job) => {
  // Children finished first; read what each one returned
  const results = await job.getChildrenValues();
  return { orderId: job.data.orderId, shipped: true, checks: results };
}, { embedded: false });

const node = await flow.add<OrderData>({
  name: 'fulfill-order',
  queueName: 'orders',
  data: { orderId: 'ORD-123' },
  children: [
    { name: 'check-inventory', queueName: 'checks', data: { orderId: 'ORD-123' } },
    { name: 'check-payment', queueName: 'checks', data: { orderId: 'ORD-123' } },
  ],
});

const result = await node.job.waitUntilFinished(null, 10_000);
console.log(result);

await checks.close();
await orders.close();
await flow.close();
```

</TabItem>
<TabItem label="Python">

```python
from bunqueue import FlowProducer, Worker

flow = FlowProducer()

flow.add({
    "name": "fulfill-order",
    "queueName": "orders",
    "data": {"orderId": "ORD-123"},
    "children": [
        {"name": "check-inventory", "queueName": "checks", "data": {"orderId": "ORD-123"}},
        {"name": "check-payment", "queueName": "checks", "data": {"orderId": "ORD-123"}},
    ],
})

def fulfill(job):
    # Children finished first; read what each one returned
    results = job.get_children_values()
    return ship_order(job.data["orderId"], results)

Worker("orders", fulfill)
```

</TabItem>
<TabItem label="PHP">

```php
use Bunqueue\FlowProducer;
use Bunqueue\Queue;
use Bunqueue\Worker;

$flow = new FlowProducer();

$flow->add([
    'name' => 'fulfill-order',
    'queueName' => 'orders',
    'data' => ['orderId' => 'ORD-123'],
    'children' => [
        ['name' => 'check-inventory', 'queueName' => 'checks', 'data' => ['orderId' => 'ORD-123']],
        ['name' => 'check-payment', 'queueName' => 'checks', 'data' => ['orderId' => 'ORD-123']],
    ],
]);

$orders = new Queue('orders');
$worker = new Worker('orders', function (Bunqueue\Job $job) use ($orders) {
    // Children finished first; read what each one returned
    $results = $orders->getChildrenValues($job->id());
    return shipOrder($job->data()['orderId'], $results);
});
$worker->run();
```

</TabItem>
<TabItem label="Go">

```go
flow := bunqueue.NewFlowProducer(bunqueue.Options{})
defer flow.Close()

flow.Add(bunqueue.FlowJob{
    Name: "fulfill-order", QueueName: "orders",
    Data: map[string]any{"orderId": "ORD-123"},
    Children: []bunqueue.FlowJob{
        {Name: "check-inventory", QueueName: "checks", Data: map[string]any{"orderId": "ORD-123"}},
        {Name: "check-payment", QueueName: "checks", Data: map[string]any{"orderId": "ORD-123"}},
    },
})

orders := bunqueue.NewQueue("orders", bunqueue.Options{})
worker := bunqueue.NewWorker("orders", func(job *bunqueue.Job) (any, error) {
    // Children finished first; read what each one returned
    results, err := orders.GetChildrenValues(job.ID())
    if err != nil {
        return nil, err
    }
    return shipOrder(job.Data()["orderId"].(string), results)
}, bunqueue.WorkerOptions{})
worker.Run()
```

</TabItem>
<TabItem label="Rust">

Not shown here. The Rust equivalent is in the [Rust SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/rust) and the [SDK guide](/guide/sdks/).

</TabItem>
<TabItem label="Elixir">

Not shown here. The Elixir equivalent is in the [Elixir SDK reference](https://github.com/egeominotti/bunqueue/tree/main/sdk/elixir) and the [SDK guide](/guide/sdks/).

</TabItem>
</Tabs>

*Rust and Elixir support the same flow trees (`FlowProducer::add` / `Bunqueue.FlowProducer.add`) and sequential chains, see the [SDK guide](/guide/sdks/#flows).*

`flow.addChain([...])` runs jobs one after another, and `flow.addBulkThen(jobs, finalJob)` fans out in parallel and merges at the end (fan-in is available in the TypeScript and Python clients). See the [flow guide](/guide/flow/). For richer orchestration (branching, rollback on failure, human approval steps) use the [workflow engine](/guide/workflow/).

## Job options cheat sheet

The options you saw above, all set per job or via `defaultJobOptions`:

| Option | What it does | Default |
| --- | --- | --- |
| `attempts` | Max tries before the job goes to the DLQ | 3 |
| `backoff` | Base delay between retries, doubles each time | 1000 ms |
| `timeout` | Max processing time before the job is failed | none |
| `priority` | Higher numbers run sooner | 0 |
| `delay` | Wait this many ms before the job is runnable | 0 |
| `jobId` | Custom ID, adding the same ID twice returns the existing job | auto |
| `durable` | SQLite: bypass its 10ms buffer; PostgreSQL is already transactional | false |
| `removeOnComplete` | Delete the job once it succeeds | false |

Full list in the [queue guide](/guide/queue/).

## Gotchas

- **No `dataPath` means no persistence.** An embedded queue without `dataPath` (or a `DATA_PATH` env var) keeps everything in memory and loses it on restart.
- **SQLite's default write buffer trades 10ms for speed.** A hard crash can lose
  jobs accepted in that window; use `durable: true` where that matters.
  PostgreSQL has no equivalent admission buffer.
- **Throwing is how you retry.** A worker that catches every error and returns normally marks the job completed. Let errors propagate when you want a retry.
- **Do not auto-retry money.** Set `setDlqConfig({ autoRetry: false })` on payment-like queues so failed charges wait for review.
- **Close workers on shutdown.** `await worker.close()` waits for active jobs to finish; `worker.close(true)` forces a stop. See [production](/guide/production/) for the full shutdown pattern.

## More patterns

- [AI agents via MCP](/guide/mcp/), let Claude or any MCP client schedule and monitor jobs
- [Multi-tenant isolation](/guide/queue-group/), one namespaced queue set per tenant
- [Rate-limited API calls](/guide/rate-limiting/), token buckets and worker limiters
- [Edge and IoT forwarding](/guide/iot-edge/), queue locally, drain to a central server
- [Copy-paste examples](/examples/), shorter recipes for common tasks

<script is:inline src="/bq-inter.js"></script>