queuebear

queuebear

QueueBear SDK for building durable workflows and managing message queues.

Installation

npm install queuebear

Quick Start

import { QueueBear, serve } from "queuebear";

const qb = new QueueBear({
  apiKey: "qb_live_xxx",
  projectId: "proj_xxx",
});

API Overview

The SDK provides access to all QueueBear APIs:

API	Description
qb.messages	Publish and manage webhook messages
qb.schedules	Create and manage cron-based recurring jobs
qb.dlq	Manage failed messages in the dead letter queue
qb.workflows	Trigger and manage durable workflows

Messages API

Publish messages to be delivered to webhook destinations with automatic retries.

Publish a Message

const { messageId } = await qb.messages.publish(
  "https://api.example.com/webhook",
  { event: "user.created", userId: "123" },
  {
    delay: "30s", // Delay before delivery
    retries: 5, // Number of retry attempts
    method: "POST", // HTTP method
    headers: { "X-API-Key": "secret" }, // Headers to forward
    callbackUrl: "https://...", // Success callback
    failureCallbackUrl: "https://...", // Failure callback
    deduplicationId: "unique-id", // Prevent duplicate messages
  }
);

Get Message Status

const message = await qb.messages.get(messageId);
console.log(message.status); // "pending" | "completed" | "failed"
console.log(message.deliveryLogs); // Delivery attempt history

List Messages

const { messages, pagination } = await qb.messages.list({
  status: "pending",
  limit: 20,
  offset: 0,
});

Cancel a Message

await qb.messages.cancel(messageId);

Publish and Wait

const message = await qb.messages.publishAndWait(
  "https://api.example.com/webhook",
  { event: "user.created" },
  { timeoutMs: 30000 }
);
console.log(message.status); // "completed"

Schedules API

Create cron-based recurring jobs.

Create a Schedule

const schedule = await qb.schedules.create({
  destination: "https://api.example.com/cron-job",
  cron: "0 9 * * *", // Daily at 9 AM
  timezone: "America/New_York",
  method: "POST",
  body: JSON.stringify({ type: "daily-report" }),
  headers: { "Content-Type": "application/json" },
  retries: 3,
  metadata: { jobName: "daily-report" },
});

Common Cron Expressions

Expression	Description
* * * * *	Every minute
0 * * * *	Every hour
0 9 * * *	Daily at 9:00 AM
0 9 * * 1-5	Weekdays at 9:00 AM
0 0 1 * *	First day of each month
0 */6 * * *	Every 6 hours

List Schedules

const { schedules } = await qb.schedules.list();

Pause / Resume

await qb.schedules.pause(scheduleId);
await qb.schedules.resume(scheduleId);

Delete a Schedule

await qb.schedules.delete(scheduleId);

Dead Letter Queue (DLQ) API

Manage messages that failed all retry attempts.

List DLQ Entries

const { entries } = await qb.dlq.list();
for (const entry of entries) {
  console.log(`${entry.id}: ${entry.failureReason}`);
}

Get Entry Details

const entry = await qb.dlq.get(dlqId);
console.log(entry.body); // Original message body
console.log(entry.totalAttempts); // Number of failed attempts

Retry a Failed Message

const result = await qb.dlq.retry(dlqId);
console.log(result.newMessageId); // New message created

Delete Entry / Purge All

await qb.dlq.delete(dlqId);
await qb.dlq.purge(); // Delete all entries

Retry All Failed Messages

const results = await qb.dlq.retryAll();
console.log(`Retried ${results.length} entries`);

Workflows API

Build durable, fault-tolerant workflows with automatic step caching.

Workflows consist of two parts:

• Workflow endpoint - Created with serve(), handles workflow execution
• Client - Uses qb.workflows to trigger and manage workflow runs

serve() - Create a Workflow Endpoint

The serve() function creates an HTTP handler for your workflow. It receives requests from QueueBear, executes your workflow code, and manages step caching automatically.

import { serve } from "queuebear";

export const POST = serve<InputType>(async (context) => {
  // Your workflow logic here
  return result;
}, options);

Parameters:

Parameter	Type	Description
handler	(context: WorkflowContext<T>) => Promise<R>	Your workflow function
options	ServeOptions	Optional configuration

Options:

Option	Type	Description
signingSecret	string	Secret to verify requests come from QueueBear

Framework Integration

Next.js (App Router)

// app/api/workflows/my-workflow/route.ts
import { serve } from "queuebear";

export const POST = serve(async (context) => {
  await context.run("step-1", async () => {
    /* ... */
  });
  return { success: true };
});

Express

import express from "express";
import { serve } from "queuebear";

const app = express();
app.use(express.json());

const handler = serve(async (context) => {
  await context.run("step-1", async () => {
    /* ... */
  });
  return { success: true };
});

app.post("/api/workflows/my-workflow", async (req, res) => {
  const response = await handler(
    new Request(req.url, {
      method: "POST",
      headers: req.headers as HeadersInit,
      body: JSON.stringify(req.body),
    })
  );
  res.status(response.status).json(await response.json());
});

Hono

import { Hono } from "hono";
import { serve } from "queuebear";

const app = new Hono();

const handler = serve(async (context) => {
  await context.run("step-1", async () => {
    /* ... */
  });
  return { success: true };
});

app.post("/api/workflows/my-workflow", async (c) => {
  const response = await handler(c.req.raw);
  return response;
});

Complete Workflow Example

// app/api/workflows/onboarding/route.ts
import { serve } from "queuebear";

interface OnboardingInput {
  userId: string;
  email: string;
}

export const POST = serve<OnboardingInput>(
  async (context) => {
    const { userId, email } = context.input;

    // Step 1: Send welcome email (cached if already done)
    await context.run("send-welcome", async () => {
      await sendEmail(email, "welcome");
    });

    // Step 2: Wait 3 days
    await context.sleep("wait-3-days", 60 * 60 * 24 * 3);

    // Step 3: Send tips email
    await context.run("send-tips", async () => {
      await sendEmail(email, "tips");
    });

    return { completed: true };
  },
  {
    signingSecret: process.env.QUEUEBEAR_SIGNING_SECRET,
  }
);

Trigger a Workflow

const { runId } = await qb.workflows.trigger(
  "user-onboarding",
  "https://your-app.com/api/workflows/onboarding",
  { userId: "123", email: "[e-mail]" },
  {
    idempotencyKey: "onboarding-user-123",
    maxDuration: 60 * 60 * 24 * 7, // 7 day timeout
  }
);

Check Workflow Status

const status = await qb.workflows.getStatus(runId);
console.log(status.status); // "running" | "sleeping" | "completed"
console.log(status.steps); // Array of step details

Wait for Completion

const result = await qb.workflows.waitForCompletion(runId, {
  pollIntervalMs: 2000,
  timeoutMs: 60000,
});

Trigger and Wait

const result = await qb.triggerAndWait(
  "user-onboarding",
  "https://your-app.com/api/workflows/onboarding",
  { userId: "123" },
  { timeoutMs: 120000 }
);
console.log(result.result); // Workflow output

Cancel / Retry

await qb.workflows.cancel(runId);
await qb.workflows.retry(runId); // Resume from last completed step

Send Events

// In workflow: await context.waitForEvent("order-approved", "order.approved")

// From external code:
await qb.workflows.sendEvent("order.approved", {
  eventKey: "order-123",
  payload: { status: "approved" },
});

Context Methods

Available in serve() handlers:

context.run(stepName, fn, options?)

Execute a step with automatic caching.

const result = await context.run("fetch-user", async () => {
  return await db.users.findById(userId);
});

context.sleep(stepName, seconds)

Pause workflow for specified duration.

await context.sleep("wait-1-hour", 3600);

context.sleepUntil(stepName, date)

Pause until a specific date/time.

await context.sleepUntil("wait-until-tomorrow", new Date("2024-01-15"));

context.call(stepName, config)

Make an HTTP call as a cached step.

const data = await context.call("fetch-api", {
  url: "https://api.example.com/data",
  method: "POST",
  headers: { Authorization: "Bearer xxx" },
  body: { key: "value" },
});

context.waitForEvent(stepName, eventName, options?)

Wait for an external event.

const payload = await context.waitForEvent("wait-approval", "order.approved", {
  eventKey: "order-123",
  timeoutSeconds: 86400, // 1 day
});

context.notify(eventName, payload?)

Send fire-and-forget event.

await context.notify("user.onboarded", { userId: "123" });

context.parallel(steps)

Execute steps in parallel.

const [user, orders, preferences] = await context.parallel([
  { name: "fetch-user", fn: () => fetchUser(userId) },
  { name: "fetch-orders", fn: () => fetchOrders(userId) },
  { name: "fetch-preferences", fn: () => fetchPreferences(userId) },
]);

context.getCompletedSteps()

Get all completed steps for debugging.

const steps = await context.getCompletedSteps();
console.log(`Completed ${steps.length} steps`);

Security

Signature Verification

Verify that workflow requests come from your QueueBear instance:

export const POST = serve(handler, {
  signingSecret: process.env.QUEUEBEAR_SIGNING_SECRET,
});

The signing secret is available in your QueueBear project settings. When configured, requests without a valid signature will be rejected with a 401 error.

Local Development

When developing locally, your webhook endpoints run on localhost which isn't accessible from QueueBear's servers. Use Tunnelmole to expose your local server - it's free and requires no signup.

Installing Tunnelmole

Linux, macOS, Windows WSL:

curl -O https://install.tunnelmole.com/t357g/install && sudo bash install

Node.js (all platforms, requires Node 16+):

npm install -g tunnelmole

Starting a Tunnel

tmole 3000
# Output: https://xxxx.tunnelmole.com is forwarding to localhost:3000

Using the Tunnel URL

// Use tunnelmole URL instead of localhost
await qb.messages.publish("https://xxxx.tunnelmole.com/api/webhooks", {
  event: "user.created",
  userId: "123"
});

// Works for workflows too
await qb.workflows.trigger(
  "onboarding",
  "https://xxxx.tunnelmole.com/api/workflows/onboarding",
  { userId: "123" }
);

Tips

• Store your tunnel URL in .env for easy switching between local and production
• Both callbackUrl and failureCallbackUrl need public URLs for local testing
• Tunnel URLs change on restart

License

MIT