Publishing Jobs

Publishing

Use typed queue instances to publish jobs with full type safety. Each queue instance has a publish method that accepts the payload and optional provider-specific options.

Best Practice: Centralized Queue Registry

Recommended (but not enforced): Define all queues in a centralized location for better organization and reusability:

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

// Define payload types
type UserEvent = { userId: string; action: string };
type EmailJob = { to: string; subject: string; body: string };
type NotificationJob = { userId: string; message: string };

// Export all queues from a single registry
export const queues = {
  userEvents: bullmqQueue<UserEvent>("user-events"),
  emails: sqsQueue<EmailJob>("emails", {
    queueUrl: "https://sqs.us-east-1.amazonaws.com/123456789/emails",
  }),
  notifications: pgbossQueue<NotificationJob>("notifications"),
} as const;

Benefits:

• Single source of truth for all queues
• Easy to import: import { queues } from './queues/registry.js'
• Consistent queue configuration across the application
• Better discoverability of available queues

Usage:

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

// Publishing
await queues.userEvents.publish({ userId: "123", action: "signup" });
await queues.emails.publish({
  to: "user@example.com",
  subject: "Welcome",
  body: "Hello!",
});

// Subscribing
export class UserEventHandler {
  @queues.userEvents.subscribe()
  async handle(payload: UserEvent) {
    console.log(`User ${payload.userId} performed ${payload.action}`);
  }
}

This pattern is completely optional - you can also define queues inline or in separate files based on your preference.

Creating typed queues

Alternatively, create typed queue instances directly where needed using factory functions:

import { bullmqQueue, sqsQueue, pgbossQueue } from "balda";

// Define payload types
type UserEvent = { userId: string; action: string };
type EmailJob = { to: string; subject: string; body: string };
type NotificationJob = { userId: string; message: string };

// Create queue instances
export const userQueue = bullmqQueue<UserEvent>("user-events");
export const emailQueue = sqsQueue<EmailJob>("emails", {
  queueUrl: "https://sqs.us-east-1.amazonaws.com/123456789/emails",
});
export const notificationQueue = pgbossQueue<NotificationJob>("notifications");

Basic publishing

// Publish to BullMQ
await userQueue.publish({ userId: "123", action: "signup" });

// Publish to SQS
await emailQueue.publish({
  to: "user@example.com",
  subject: "Welcome",
  body: "Thanks for signing up!",
});

// Publish to PGBoss
await notificationQueue.publish({
  userId: "123",
  message: "Your order has shipped",
});

Publishing with provider options

Each provider supports specific options as the second parameter:

BullMQ options (maps to Queue.add third parameter):

await userQueue.publish(
  { userId: "123", action: "signup" },
  {
    delay: 5000, // Delay 5 seconds
    priority: 1, // High priority
    removeOnComplete: true, // Auto-remove on completion
    attempts: 3, // Retry up to 3 times
    backoff: {
      type: "exponential",
      delay: 2000,
    },
  },
);

SQS options:

await emailQueue.publish(
  { to: "user@example.com", subject: "Hello", body: "World" },
  {
    MessageGroupId: "emails", // For FIFO queues
    MessageDeduplicationId: "123", // For FIFO queues
    DelaySeconds: 10, // Delay 10 seconds
    MessageAttributes: {
      // Custom attributes
      priority: {
        DataType: "String",
        StringValue: "high",
      },
    },
  },
);

PGBoss options (maps to send third parameter):

await notificationQueue.publish(
  { userId: "123", message: "Hello!" },
  {
    retryLimit: 3, // Retry up to 3 times
    retryDelay: 60, // Wait 60 seconds between retries
    expireInSeconds: 3600, // Expire after 1 hour
    priority: 10, // Higher number = higher priority
  },
);

Subscribing to queues

The subscribe method supports two patterns:

1. Decorator-based subscription (recommended for class methods):

export class UserEventHandler {
  @userQueue.subscribe()
  async handle(payload: UserEvent) {
    console.log(`User ${payload.userId} performed ${payload.action}`);
  }
}

2. Callback-based subscription (recommended for standalone functions):

await userQueue.subscribe(async (payload) => {
  console.log(`User ${payload.userId} performed ${payload.action}`);
});

Key differences:

• Decorators are ideal for organized class-based handlers and work seamlessly with dependency injection patterns
• Callbacks are perfect for simple handlers, scripts, or when you prefer a functional programming style
• Both methods provide full type safety and identical runtime behavior
• You can mix both approaches in the same application

Type safety benefits

The typed queue API provides full type safety:

// ✅ Type-safe: correct payload
await userQueue.publish({ userId: '123', action: 'signup' });

// ❌ TypeScript error: missing required fields
await userQueue.publish({ userId: '123' });

// ❌ TypeScript error: wrong field type
await userQueue.publish({ userId: 123, action: 'signup' });

// ✅ Type-safe handler
@userQueue.subscribe()
async handle(payload: UserEvent) {
  // payload.userId is string
  // payload.action is string
  // Full autocomplete and type checking
}

Multiple queues with the same provider

You can create multiple queue instances for the same provider:

import { bullmqQueue } from 'balda';

// Different queues for different job types
export const emailQueue = bullmqQueue<EmailJob>('emails');
export const smsQueue = bullmqQueue<SMSJob>('sms');
export const pushQueue = bullmqQueue<PushJob>('push-notifications');

// Each has its own type-safe publish/subscribe
await emailQueue.publish({ to: 'user@example.com', ... });
await smsQueue.publish({ phone: '+1234567890', ... });
await pushQueue.publish({ deviceToken: 'abc123', ... });

Summary: Two ways to subscribe

Every typed queue instance (BullMQ, SQS, PGBoss, or custom) supports both subscription methods:

import { bullmqQueue, sqsQueue, pgbossQueue } from "balda";

const queue = bullmqQueue<MyPayload>("my-topic");

// Method 1: Decorator (for class methods)
class MyHandler {
  @queue.subscribe()
  async handle(payload: MyPayload) {}
}

// Method 2: Callback (for standalone functions)
await queue.subscribe(async (payload) => {});

Choose the method that best fits your application architecture. Both provide identical functionality and full TypeScript type safety.