Queueflow

A simple, type-safe task queue library that gives you the developer-friendly API of Trigger.dev or Inngest, but built on top of BullMQ and Redis. No infrastructure to manage, no platform lock-in—just define your tasks and trigger them.

If you've ever used Trigger.dev or Inngest, you know how nice their APIs are. They're clean, type-safe, and make background jobs feel effortless. But what if you want that same experience while keeping control of your infrastructure?

That's where Queueflow comes in. I couldn't find an npm package that combined BullMQ's reliability with a Trigger.dev-like API, so I built one. It's perfect for when you want:

• ✅ A simple, type-safe API (like Trigger.dev/Inngest)
• ✅ Full control over your Redis/BullMQ infrastructure
• ✅ Zero boilerplate—just define and trigger
• ✅ TypeScript-first with full type safety

bun add @benrobo/queueflow

Or with npm:

npm install @benrobo/queueflow

You'll also need Redis running (locally or remote). If you don't have it set up yet, you can run it with Docker:

docker run -d -p 6379:6379 redis:latest

// tasks/email.ts
import { defineTask } from "@benrobo/queueflow";

export const sendWelcomeEmail = defineTask({
  id: "email.welcome",
  run: async (payload: { userId: string; email: string }) => {
    console.log(`Sending welcome email to ${payload.email}`);
    // Your email sending logic here
  },
});

If you're using a local Redis instance on the default port, you can skip this step. Otherwise, configure it once at your app's entry point. The configure() function validates the Redis connection immediately and will throw an error if the connection fails:

// app.ts or index.ts
import { configure } from "@benrobo/queueflow";

// Using await (recommended)
await configure({
  connection: process.env.REDIS_URL || "redis://localhost:6379",
  defaultQueue: "myapp",
});

// Or with error handling using .catch()
configure({
  connection: process.env.REDIS_URL || "redis://localhost:6379",
  defaultQueue: "myapp",
}).catch((error) => {
  console.error("Failed to configure Queueflow:", error);
  process.exit(1);
});

You can use either a connection string or an object:

await configure({
  connection: {
    host: "localhost",
    port: 6379,
    password: "your-password",
    db: 0,
  },
});

Enable debug logging to see detailed information about task registration, triggering, and processing:

await configure({
  connection: process.env.REDIS_URL,
  debug: true, // Enable debug logs
});

import { sendWelcomeEmail } from "./tasks/email";

await sendWelcomeEmail.trigger({
  userId: "123",
  email: "user@example.com",
});

That's it! The worker automatically starts when you trigger your first task.

Queueflow uses a singleton worker pattern that automatically manages everything for you:

1. Define tasks with defineTask() - they register themselves automatically
2. Trigger tasks with .trigger() - the worker starts on first use
3. That's it - no worker setup, no queue configuration, no boilerplate

The worker listens to all your queues in the background and processes jobs as they come in. Tasks are organized by queue (defaults to the prefix of your task ID, like email from email.welcome).

Queueflow is built with TypeScript from the ground up. Your payload types flow through from definition to trigger:

const processOrder = defineTask({
  id: "orders.process",
  run: async (payload: { orderId: string; amount: number }) => {
    // payload is fully typed here
    console.log(`Processing order ${payload.orderId}`);
  },
});

// TypeScript will catch this error:
await processOrder.trigger({
  orderId: "123",
  // amount: 100, // ❌ TypeScript error: missing 'amount'
});

Need to schedule a task for later? Pass options as the second parameter to trigger:

await sendWelcomeEmail.trigger(
  { userId: "123", email: "user@example.com" },
  { delay: 5000 } // Send in 5 seconds
);

You can pass any BullMQ job options through the second parameter, including:

• delay - milliseconds to wait before processing
• attempts - number of retry attempts
• backoff - retry strategy

Handle errors gracefully with the onError callback:

const sendEmail = defineTask({
  id: "email.send",
  run: async (payload: { email: string; subject: string }) => {
    await sendEmailToUser(payload.email, payload.subject);
  },
  onError: async (error, payload) => {
    console.error(`Failed to send email to ${payload.email}:`, error.message);
    await logErrorToService(error, payload);
  },
});

The onError callback receives both the error and the original payload, so you can handle failures appropriately (log to an error service, send notifications, etc.).

For recurring tasks that run on a schedule, use scheduleTask:

import { scheduleTask } from "@benrobo/queueflow";

const dailyReport = scheduleTask({
  id: "reports.daily",
  cron: "0 9 * * *",
  run: async () => {
    console.log("Generating daily report...");
    await generateReport();
  },
  onError: async (error) => {
    console.error("Failed to generate report:", error.message);
  },
});

The task will automatically run based on the cron pattern. Common cron patterns:

• "0 9 * * *" - Every day at 9:00 AM
• "0 * * * *" - Every hour
• "0 0 * * 0" - Every Sunday at midnight
• "*/5 * * * *" - Every 5 minutes

You can also specify a timezone:

const weeklyBackup = scheduleTask({
  id: "backup.weekly",
  cron: "0 2 * * 0",
  tz: "America/New_York",
  run: async () => {
    await performBackup();
  },
});

By default, tasks are grouped by the prefix of their ID (e.g., email.welcome goes to the email queue). You can override this:

const heavyTask = defineTask({
  id: "processing.heavy",
  queue: "heavy-processing", // Custom queue name
  run: async (payload) => {
    // This runs in the "heavy-processing" queue
  },
});

Control how many tasks run simultaneously per queue:

const highVolumeTask = defineTask({
  id: "processing.high-volume",
  concurrency: 10,
  run: async (payload) => {
    // This queue can process up to 10 tasks concurrently
  },
});

While the worker auto-starts, you can also start it manually if you prefer:

import { startWorker } from "@benrobo/queueflow";

startWorker();

This is useful if you want to ensure the worker is running before your app starts accepting requests.

Here's a more complete example showing how you might structure tasks in a real app:

// tasks/email.ts
import { defineTask } from "@benrobo/queueflow";

export const sendWelcomeEmail = defineTask({
  id: "email.welcome",
  run: async (payload: { userId: string; email: string }) => {
    await sendEmail({
      to: payload.email,
      subject: "Welcome!",
      body: "Thanks for joining us!",
    });
  },
});

export const sendPasswordReset = defineTask({
  id: "email.password-reset",
  run: async (payload: { email: string; token: string }) => {
    await sendEmail({
      to: payload.email,
      subject: "Reset your password",
      body: `Click here to reset: ${payload.token}`,
    });
  },
});

// tasks/orders.ts
import { defineTask } from "@benrobo/queueflow";

export const processOrder = defineTask({
  id: "orders.process",
  run: async (payload: { orderId: string; items: string[] }) => {
    // Process the order
    await chargePayment(payload.orderId);
    await updateInventory(payload.items);
    await sendConfirmation(payload.orderId);
  },
});

// app.ts
import { configure } from "@benrobo/queueflow";
import "./tasks/email";
import "./tasks/orders";

await configure({
  connection: process.env.REDIS_URL,
});

// Later in your API routes or wherever:
import { sendWelcomeEmail, processOrder } from "./tasks";

app.post("/signup", async (req, res) => {
  const user = await createUser(req.body);

  await sendWelcomeEmail.trigger({
    userId: user.id,
    email: user.email,
  });

  res.json({ success: true });
});

Creates a new task definition for manual triggering.

Parameters:

• config.id (string) - Unique identifier for the task (e.g., "email.welcome")
• config.queue (string, optional) - Queue name (defaults to defaultQueue from config)
• config.handler (function) - Async function that processes the task payload
• config.onError (function, optional) - Error handler callback (error: Error, payload: T) => Promise<void> | void
• config.concurrency (number, optional) - Number of concurrent tasks for this queue (defaults to 5)

Returns: Task<T> instance

Creates a scheduled task that runs automatically based on a cron pattern.

Parameters:

• config.id (string) - Unique identifier for the task (e.g., "reports.daily")
• config.cron (string) - Cron pattern (e.g., "0 9 * * *" for daily at 9 AM)
• config.queue (string, optional) - Queue name (defaults to defaultQueue from config)
• config.handler (function) - Async function that processes the task
• config.onError (function, optional) - Error handler callback (error: Error, payload: T) => Promise<void> | void
• config.tz (string, optional) - Timezone (e.g., "America/New_York")
• config.concurrency (number, optional) - Number of concurrent tasks for this queue (defaults to 5)

Returns: ScheduledTask<T> instance

Note: Scheduled tasks start automatically when the worker starts. No need to call .trigger().

Triggers a task to run. Options are optional and can be used for delayed jobs, retries, and more.

Parameters:

• payload (T) - The typed payload for the task
• options (JobsOptions, optional) - BullMQ job options (delay, attempts, backoff, etc.)

Returns: Promise<void>

Configures the Redis connection and default settings. Validates the Redis connection immediately and throws an error if the connection fails.

Parameters:

• config.connection (string | object, optional) - Redis connection string or config object
• config.defaultQueue (string, optional) - Default queue name (defaults to "default")
• config.debug (boolean, optional) - Enable debug logging (defaults to false)

Returns: Promise<void>

Throws: Error if Redis connection cannot be established

Example:

// With await
await configure({
  connection: process.env.REDIS_URL,
  defaultQueue: "myapp",
});

// With error handling
configure({
  connection: process.env.REDIS_URL,
}).catch((error) => {
  console.error("Configuration failed:", error);
  process.exit(1);
});

// With debug logging enabled
await configure({
  connection: process.env.REDIS_URL,
  debug: true,
});

Manually starts the global worker. Usually not needed as it auto-starts on first trigger.

• Node.js 18+ or Bun
• Redis (local or remote)
• TypeScript 5.0+ (for type safety)

MIT

Contributions are welcome! This is a young project, so I'm open to ideas, bug reports, and pull requests.