Bun 队列

Install

bun add bunqueue

Requires Bun runtime. Node.js is not supported.

Quickstart

bun add bunqueue

import { Bunqueue } from 'bunqueue/client';

const app = new Bunqueue('emails', {
  embedded: true,
  processor: async (job) => {
    console.log(`Sending to ${job.data.to}`);
    return { sent: true };
  },
});

await app.add('send', { to: 'alice@example.com' });

That's it. Queue + Worker in one object. No Redis, no config, no setup.

---

Full Example

import { Bunqueue, shutdownManager } from 'bunqueue/client';

const app = new Bunqueue<{ payload: string }>('my-app', {
  embedded: true,
  routes: {
    'process': async (job) => ({ id: job.data.payload, status: 'done' }),
    'notify': async (job) => ({ sent: true }),
    'alert': async (job) => ({ alerted: true }),
  },
  concurrency: 10,
  retry: { maxAttempts: 3, delay: 1000, strategy: 'jitter' },
  circuitBreaker: { threshold: 5, resetTimeout: 30000 },
  ttl: { defaultTtl: 600000, perName: { 'verify-otp': 60000 } },
  priorityAging: { interval: 60000, minAge: 300000, boost: 1 },
  deduplication: { ttl: 5000 },
  rateLimit: { max: 100, duration: 1000 },
  dlq: { autoRetry: true, maxAge: 604800000 },
});

app.use(async (job, next) => {
  const start = Date.now();
  const result = await next();
  console.log(`${job.name}: ${Date.now() - start}ms`);
  return result;
});

app
  .trigger({ on: 'process', create: 'notify', data: (r) => ({ payload: r.id }) })
  .trigger({ on: 'process', event: 'failed', create: 'alert', data: (_, j) => j.data });

await app.cron('cleanup', '0 2 * * *', { payload: 'nightly' });
await app.add('process', { payload: 'ORD-001' });

process.on('SIGINT', async () => {
  await app.close();
  shutdownManager();
});

Simple Mode docs →

---

Workflow Engine

Orchestrate multi-step business processes with branching, saga compensation, and human-in-the-loop signals. Built on top of bunqueue — no new infrastructure.

import { Workflow, Engine } from 'bunqueue/workflow';

const orderFlow = new Workflow('order-pipeline')
  .step('validate', async (ctx) => {
    const { orderId, amount } = ctx.input as { orderId: string; amount: number };
    if (amount <= 0) throw new Error('Invalid amount');
    return { orderId };
  })
  .step('reserve-stock', async () => {
    await inventory.reserve();
    return { reserved: true };
  }, {
    compensate: async () => await inventory.release(), // Auto-rollback on failure
  })
  .step('charge', async () => {
    return { txId: await payments.charge() };
  }, {
    compensate: async () => await payments.refund(),
  })
  .step('confirm', async (ctx) => {
    const { txId } = ctx.steps['charge'] as { txId: string };
    return { emailSent: true, txId };
  });

const engine = new Engine({ embedded: true });
engine.register(orderFlow);
await engine.start('order-pipeline', { orderId: 'ORD-1', amount: 99.99 });

Features:

• Saga pattern — Compensation handlers run in reverse when a step fails
• Branching — Route to different paths based on runtime conditions
• Parallel steps — Run independent steps concurrently with .parallel()
• Human-in-the-loop — waitFor('event') pauses execution, engine.signal() resumes it
• Signal timeout — waitFor('event', { timeout }) fails if signal doesn't arrive in time
• Step retry — Automatic retry with exponential backoff and jitter
• Nested workflows — Compose workflows with .subWorkflow(), child results passed back
• Loops — doUntil() and doWhile() for conditional iteration with safety limits
• forEach — Iterate over dynamic item lists with indexed step results (step:0, step:1, ...)
• Map — Synchronous data transforms between steps with .map()
• Schema validation — Validate step input/output with Zod, ArkType, Valibot, or any .parse() schema
• Subscribe — engine.subscribe(id, callback) to monitor a specific execution's events
• Observability — Typed event emitter with 11 event types (engine.on/onAny)
• Cleanup & archival — engine.cleanup() / engine.archive() for execution history management
• Step timeouts — Prevent steps from running indefinitely
• Context passing — Each step accesses input and all previous step results
• SQLite persistence — Execution state survives restarts

vs Competitors:

// Branching
const flow = new Workflow('tiered')
  .step('classify', async (ctx) => {
    const { amount } = ctx.input as { amount: number };
    return { tier: amount > 1000 ? 'vip' : 'basic' };
  })
  .branch((ctx) => (ctx.steps['classify'] as { tier: string }).tier)
  .path('vip', (w) => w.step('vip-handler', async () => ({ discount: 20 })))
  .path('basic', (w) => w.step('basic-handler', async () => ({ discount: 0 })))
  .step('done', async () => ({ processed: true }));

// Human-in-the-loop
const approvalFlow = new Workflow('expense')
  .step('submit', async (ctx) => {
    const { amount } = ctx.input as { amount: number };
    return { amount };
  })
  .waitFor('manager-approval')
  .step('reimburse', async (ctx) => {
    const decision = ctx.signals['manager-approval'] as { approved: boolean };
    return { status: decision.approved ? 'paid' : 'rejected' };
  });

// Signal a waiting workflow
await engine.signal(run.id, 'manager-approval', { approved: true });

Workflow Engine docs →

---

<p align="center"> <strong>bunqueue Dashboard</strong><br/> _{A visual interface for managing queues, jobs, workers and monitoring in real time. Currently in beta.} </p>

https://github.com/user-attachments/assets/e8a8d38e-b4a6-4dc8-8360-876c0f24d116

<p align="center"> _{Want early access? Reach out at <b>egeominotti@gmail.com</b>} </p>

---