Queue Providers

Providers

• Balda Queue supports the following providers out of the box:

  • BullMQ
  • SQS
  • PgBoss

• You must also install the corresponding package for each provider:

  • BullMQ: npm install bullmq ioredis
  • SQS: npm install @aws-sdk/client-sqs sqs-consumer
  • PgBoss: npm install pg-boss pg

• It's also possible to support custom providers via the GenericPubSub interface.

Single entrypoint configuration

Use defineQueueConfiguration to configure all providers in one place during app bootstrap:

import { defineQueueConfiguration } from "balda";

defineQueueConfiguration({
  bullmq: {
    connection: {
      host: "127.0.0.1",
      username: "default",
      password: "root",
      db: 0,
    },
    defaultJobOptions: { removeOnComplete: 100, removeOnFail: 100 },
  },
  sqs: {
    client: { region: "us-east-1" },
    consumer: {
      queueUrlMap: {
        test: "http://localhost:9324/queue/balda-development-test",
      },
      batchSize: 10,
    },
  },
  pgboss: {
    connectionString: process.env.DATABASE_URL!,
  },
});

Best Practice: Centralized Queue Registry

While not enforced, we recommend creating a centralized registry for all your queues. This provides a single source of truth and makes queue instances easily accessible throughout your application:

// src/queues/index.ts
import { bullmqQueue, sqsQueue, pgbossQueue } from "balda";

// Payload type definitions
type EmailPayload = { to: string; subject: string; body: string };
type OrderPayload = { orderId: number; total: number };
type NotificationPayload = { userId: string; message: string };

// Centralized queue registry
export const queues = {
  // BullMQ queues
  emails: bullmqQueue<EmailPayload>("emails"),

  // SQS queues
  orders: sqsQueue<OrderPayload>("orders", {
    queueUrl: "https://sqs.us-east-1.amazonaws.com/123456789/orders",
  }),

  // PGBoss queues
  notifications: pgbossQueue<NotificationPayload>("notifications"),
} as const;

Then use throughout your application:

import { queues } from "./queues/index.js";

// In handlers
export class EmailHandler {
  @queues.emails.subscribe()
  async handle(payload: EmailPayload) {
    // Send email
  }
}

// In services
await queues.emails.publish({
  to: "user@example.com",
  subject: "Welcome",
  body: "Hello!",
});

BullMQ

Create a typed BullMQ queue and subscribe to it using either decorators or callbacks:

import { bullmqQueue } from "balda";

type EmailPayload = {
  to: string;
  subject: string;
  body: string;
};

// Create queue instance
export const emailQueue = bullmqQueue<EmailPayload>("emails");

// Subscribe with decorator (recommended for class methods)
export class EmailHandler {
  @emailQueue.subscribe()
  async handle(payload: EmailPayload) {
    console.log(`Sending email to ${payload.to}`);
  }
}

// Or subscribe with callback (recommended for standalone functions)
await emailQueue.subscribe(async (payload) => {
  console.log(`Sending email to ${payload.to}`);
});

// Publish
await emailQueue.publish({
  to: "user@example.com",
  subject: "Hello",
  body: "World",
});

SQS

Create a typed SQS queue with queue URL and subscribe using either method:

import { sqsQueue } from "balda";

type OrderPayload = {
  orderId: number;
  total: number;
};

// Create queue instance with SQS options
export const orderQueue = sqsQueue<OrderPayload>("orders", {
  queueUrl: "https://sqs.us-east-1.amazonaws.com/123456789/my-queue",
});

// Subscribe with decorator
export class OrderHandler {
  @orderQueue.subscribe()
  async handle(payload: OrderPayload) {
    console.log(`Processing order ${payload.orderId}`);
  }
}

// Or subscribe with callback
await orderQueue.subscribe(async (payload) => {
  console.log(`Processing order ${payload.orderId}`);
});

// Publish with SQS-specific options
await orderQueue.publish(
  { orderId: 123, total: 99.99 },
  {
    MessageGroupId: "orders",
    DelaySeconds: 10,
  },
);

Note: You can either specify queueUrl per queue instance (as shown above) or configure it globally in defineQueueConfiguration with queueUrlMap.

PGBoss

Create a typed PGBoss queue and subscribe using either method:

import { pgbossQueue } from "balda";

type NotificationPayload = {
  userId: string;
  message: string;
};

// Create queue instance
export const notificationQueue =
  pgbossQueue<NotificationPayload>("notifications");

// Subscribe with decorator
export class NotificationHandler {
  @notificationQueue.subscribe()
  async handle(payload: NotificationPayload) {
    console.log(`Notifying user ${payload.userId}: ${payload.message}`);
  }
}

// Or subscribe with callback
await notificationQueue.subscribe(async (payload) => {
  console.log(`Notifying user ${payload.userId}: ${payload.message}`);
});

// Publish with PGBoss options
await notificationQueue.publish(
  { userId: "123", message: "Hello!" },
  {
    retryLimit: 3,
    retryDelay: 60,
  },
);

Note: PgBoss automatically creates queues when needed. Ensure your PostgreSQL database is accessible and the pg-boss package is installed.

Custom provider

Implement the GenericPubSub interface and create a typed queue using createQueue:

import { createQueue, GenericPubSub } from "balda";

type CustomPayload = {
  data: string;
};

// Implement custom PubSub
class MyCustomPubSub implements GenericPubSub<CustomPayload> {
  async publish(
    topic: string,
    payload: CustomPayload,
  ): Promise<{ id: string }> {
    // Custom publish logic
    console.log("Publishing to", topic, payload);
    return { id: "custom-id-1" };
  }

  async subscribe(
    topic: string,
    handler: (payload: CustomPayload) => Promise<void>,
  ): Promise<void> {
    // Wire your consumer and call handler(payload)
    console.log("Subscribed to", topic);
  }
}

// Create typed queue with custom provider
export const customQueue = createQueue<CustomPayload>(
  "custom-topic",
  new MyCustomPubSub(),
);

// Subscribe with decorator
export class CustomHandler {
  @customQueue.subscribe()
  async handle(payload: CustomPayload) {
    console.log("Received:", payload.data);
  }
}

// Or subscribe with callback
await customQueue.subscribe(async (payload) => {
  console.log("Received:", payload.data);
});

// Publish
await customQueue.publish({ data: "test" });

The custom provider approach gives you full control over the queue implementation while maintaining type safety and a consistent API.