Building Event-Driven Microservices with BeeQueue: A Complete Architecture Guide

If you're building microservices in Node.js, you've probably felt the pain of synchronous HTTP communication between services — tight coupling, cascading failures, timeout soup, and services that can't be deployed independently. Message brokers like RabbitMQ and Kafka solve these problems, but they introduce significant infrastructure complexity and operational overhead.

BeeQueue offers a compelling middle ground. At roughly 1,000 lines of code with Redis as its only dependency, it provides a lightweight but production-proven backbone for event-driven microservices. With built-in Pub/Sub events, retries with backoff, at-least-once delivery guarantees, and a simple API, BeeQueue can serve as your inter-service communication layer without the overhead of a dedicated message broker.

This guide covers real-world production patterns for using BeeQueue as the backbone of an event-driven microservices architecture — from communication patterns and graceful shutdown in Kubernetes to dead letter queues, circuit breakers, testing strategies, and framework integration.

Section 1: Microservice Communication Patterns

The fundamental challenge of microservices is getting services to talk to each other reliably. BeeQueue supports three core patterns that replace synchronous HTTP calls with async, resilient messaging.

Request/Reply Pattern

Sometimes you need a synchronous-feeling call — "process this payment and give me the result" — but you want the reliability guarantees that a queue provides. BeeQueue's job event system makes this straightforward:

import Queue from "bee-queue";

interface RequestReplyOptions {
  queueName: string;
  timeout?: number;
  retries?: number;
  redisUrl?: string;
}

export async function requestReply<TReq, TRes>(
  options: RequestReplyOptions,
  payload: TReq,
): Promise<TRes> {
  const { queueName, timeout = 30000, retries = 0, redisUrl } = options;

  const queue = new Queue(queueName, {
    redis: redisUrl ?? process.env.REDIS_URL ?? "redis://127.0.0.1:6379",
    isWorker: false,
  });

  const job = queue
    .createJob(payload as unknown as Record<string, unknown>)
    .retries(retries)
    .timeout(timeout);

  await job.save();

  return new Promise<TRes>((resolve, reject) => {
    const timer = setTimeout(() => {
      job.removeListener("succeeded", onSuccess);
      job.removeListener("failed", onFailed);
      reject(new Error(`Request/reply timed out after ${timeout}ms`));
    }, timeout);

    function onSuccess(result: TRes) {
      clearTimeout(timer);
      resolve(result);
    }

    function onFailed(err: Error) {
      clearTimeout(timer);
      reject(err);
    }

    job.on("succeeded", onSuccess);
    job.on("failed", onFailed);
  });
}

On the worker side, the handler is a plain BeeQueue processor — its return value flows back to the producer via the succeeded event:

// services/payment-service/worker.ts
const queue = new Queue("payment-processing", {
  redis: process.env.REDIS_URL!,
  isWorker: true,
});

queue.process(async (job) => {
  const { orderId, amount, currency } = job.data as {
    orderId: string; amount: number; currency: string;
  };

  const result = await processWithStripe(orderId, amount, currency);
  return { transactionId: result.id, status: "success", amount };
});

Publish/Subscribe via Queue Events

BeeQueue provides built-in Pub/Sub that broadcasts across all queue instances sharing the same Redis. This enables a lightweight event bus between services:

// lib/event-bus.ts
import Queue from "bee-queue";

type EventHandler = (jobId: string, data: unknown) => void;

export class BeeQueueEventBus {
  private queue: Queue;
  private handlers = new Map<string, EventHandler[]>();
  private listening = false;

  constructor(queueName = "event-bus", redisUrl?: string) {
    this.queue = new Queue(queueName, {
      redis: redisUrl ?? process.env.REDIS_URL ?? "redis://127.0.0.1:6379",
      isWorker: false,
    });
  }

  async publish(eventType: string, payload: unknown): Promise<string> {
    const job = this.queue.createJob({
      eventType,
      payload,
      timestamp: Date.now(),
    } as unknown as Record<string, unknown>);
    await job.save();
    return job.id;
  }

  on(eventType: string, handler: EventHandler): void {
    const existing = this.handlers.get(eventType) ?? [];
    existing.push(handler);
    this.handlers.set(eventType, existing);
    this.startListening();
  }

  private startListening(): void {
    if (this.listening) return;
    this.listening = true;

    this.queue.on("succeeded", (jobId: string, data: unknown) => {
      const result = data as { eventType: string; payload: unknown };
      const handlers = this.handlers.get(result.eventType);
      if (handlers) {
        for (const handler of handlers) {
          handler(jobId, result.payload);
        }
      }
    });
  }
}

Services can then publish and subscribe without knowing about each other:

// services/order-service/publisher.ts
const eventBus = new BeeQueueEventBus();
await eventBus.publish("order.created", { orderId, customerId, items });

// services/inventory-service/subscriber.ts
const eventBus = new BeeQueueEventBus();
eventBus.on("order.created", async (jobId, payload) => {
  const order = payload as { items: Array<{ sku: string; quantity: number }> };
  for (const item of order.items) {
    await reserveInventory(item.sku, item.quantity);
  }
});

Job Chaining (Saga Pattern)

A saga is a sequence of local transactions where each step publishes a message that triggers the next. BeeQueue's simplicity makes this natural — each worker processes a job, then creates a new job on the next service's queue:

// services/payment-service/saga-worker.ts
import Queue from "bee-queue";

const paymentQueue = new Queue("payment-process", { redis: process.env.REDIS_URL!, isWorker: true });
const inventoryQueue = new Queue("inventory-reserve", { redis: process.env.REDIS_URL!, isWorker: false });
const compensationQueue = new Queue("saga-compensate", { redis: process.env.REDIS_URL!, isWorker: false });

paymentQueue.process(async (job) => {
  const { saga } = job.data as { saga: SagaContext };

  try {
    const txnId = `txn_${Date.now()}`;
    saga.paymentTxnId = txnId;

    // Chain to next step: reserve inventory
    await inventoryQueue.createJob({ sagaAction: "reserve-inventory", saga }).save();
    return { step: "payment", status: "completed", txnId };
  } catch (err) {
    // Emit compensation
    await compensationQueue.createJob({ compensationAction: "payment-failed", saga, error: String(err) }).save();
    throw err;
  }
});

Each step follows the same pattern — process, chain forward on success, emit compensation on failure. A dedicated compensation service rolls back any completed steps when a saga fails.

Section 2: Graceful Shutdown and Process Lifecycle

In Kubernetes, pods are stopped, scaled, and restarted constantly. Your BeeQueue workers need to handle this without dropping jobs.

SIGTERM Handling

When Kubernetes sends SIGTERM, your worker must stop accepting new jobs, finish in-flight jobs, and close Redis connections cleanly:

// lib/graceful-shutdown.ts
import Queue from "bee-queue";
import { Server } from "http";

export function setupGracefulShutdown(config: {
  queues: Queue[];
  healthCheckServer?: Server;
  timeout?: number;
  appName?: string;
}): void {
  const { queues, healthCheckServer, timeout = 30000, appName = "beequeue-worker" } = config;
  let shuttingDown = false;

  async function shutdown(signal: string): Promise<void> {
    if (shuttingDown) { process.exit(1); }
    shuttingDown = true;
    console.log(`[${appName}] Received ${signal}. Shutting down...`);

    // Close queues — BeeQueue's close() finishes in-flight handlers
    await Promise.race([
      Promise.all(queues.map((q) => q.close().catch(() => {}))),
      new Promise((_, reject) => setTimeout(() => reject(new Error("timeout")), timeout)),
    ]).catch(() => {});

    console.log(`[${appName}] Shutdown complete.`);
    process.exit(0);
  }

  process.on("SIGTERM", () => shutdown("SIGTERM"));
  process.on("SIGINT", () => shutdown("SIGINT"));
}

Kubernetes Integration

Pair the shutdown handler with proper Kubernetes probes and lifecycle hooks:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-worker
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  template:
    spec:
      terminationGracePeriodSeconds: 45
      containers:
        - name: worker
          image: myapp/order-worker:latest
          ports:
            - containerPort: 9090
          lifecycle:
            preStop:
              exec:
                command: ["/bin/sh", "-c", "kill -TERM 1 && sleep 35"]
          livenessProbe:
            httpGet: { path: /healthz, port: 9090 }
            initialDelaySeconds: 5
            periodSeconds: 10
          readinessProbe:
            httpGet: { path: /readyz, port: 9090 }
            initialDelaySeconds: 3
            periodSeconds: 5

The readiness probe returns 503 during shutdown to remove the pod from service discovery, while the preStop hook sends SIGTERM and gives the worker time to drain in-flight jobs. BeeQueue's built-in stalled job detection provides the safety net — if a worker crashes mid-job, another worker re-enqueues it after the stallInterval.

Section 3: Error Handling Architecture

Building a fault-tolerant system means handling failures at every layer. Here's how to build a robust error handling stack on top of BeeQueue.

Dead Letter Queues

When a job exhausts its retries, it should land in a dead letter queue (DLQ) for inspection rather than being silently lost:

// lib/dead-letter-queue.ts
import Queue from "bee-queue";

export function setupDeadLetterQueue(config: {
  sourceQueue: Queue;
  sourceQueueName: string;
  dlqQueueName?: string;
  maxRetries: number;
}): Queue {
  const { sourceQueue, sourceQueueName, dlqQueueName = `${sourceQueueName}-dlq`, maxRetries } = config;

  const dlqQueue = new Queue(dlqQueueName, {
    redis: process.env.REDIS_URL!,
    getEvents: true,
    sendEvents: true,
  });

  sourceQueue.on("failed", async (jobId: string, err: Error) => {
    const job = await sourceQueue.getJob(jobId);
    if (!job) return;

    // Only move jobs that have exhausted retries
    if (job.status === "failed") {
      await dlqQueue.createJob({
        originalQueue: sourceQueueName,
        originalJobId: job.id,
        originalData: job.data,
        error: { message: err.message, stack: err.stack, timestamp: new Date().toISOString() },
        failedAt: Date.now(),
      } as unknown as Record<string, unknown>).save();

      console.log(`[DLQ] Moved job ${jobId} to "${dlqQueueName}"`);
    }
  });

  return dlqQueue;
}

DLQ jobs can be inspected and replayed manually — or through a dashboard like QueueHub that provides one-click retry for failed jobs.

Circuit Breaker Pattern

When a worker calls an external service (payment gateway, email API, database), a circuit breaker prevents cascading failures:

// lib/circuit-breaker.ts
import CircuitBreaker from "opossum";

export function withCircuitBreaker<TData, TResult>(
  handler: (job: import("bee-queue").Job<TData>) => Promise<TResult>,
  options: { timeout?: number; errorThresholdPercentage?: number; resetTimeout?: number; name?: string } = {},
) {
  const breaker = new CircuitBreaker(
    async (job: import("bee-queue").Job<TData>) => handler(job),
    {
      timeout: options.timeout ?? 10_000,
      errorThresholdPercentage: options.errorThresholdPercentage ?? 50,
      resetTimeout: options.resetTimeout ?? 30_000,
      name: options.name ?? "unnamed-breaker",
    },
  );

  breaker.on("open", () => console.log(`[Breaker:${breaker.name}] Circuit opened`));
  breaker.on("halfOpen", () => console.log(`[Breaker:${breaker.name}] Circuit half-open`));
  breaker.on("close", () => console.log(`[Breaker:${breaker.name}] Circuit closed`));

  return async (job: import("bee-queue").Job<TData>) => breaker.fire(job);
}

// Usage:
queue.process(withCircuitBreaker(async (job) => {
  const { orderId, amount } = job.data;
  return await paymentGateway.charge(orderId, amount);
}, { name: "payment-gateway", timeout: 5000 }));

When the circuit is open, BeeQueue's retry mechanism naturally pairs with it — the job fails fast, BeeQueue retries later, and by the next retry the circuit may have recovered.

Idempotency for Safe Retries

BeeQueue provides at-least-once delivery, which means the same job can be processed multiple times. Use idempotency keys to make retries safe:

// lib/idempotency.ts
import { createClient } from "redis";

const redis = createClient({ url: process.env.REDIS_URL! });

export async function deduplicate(
  key: string,
  ttlSeconds = 86400,
): Promise<boolean> {
  // SET NX — only succeeds if the key doesn't exist
  const result = await redis.set(key, "1", { NX: true, EX: ttlSeconds });
  return result !== null; // true = first time processing
}

// In your worker:
queue.process(async (job) => {
  const { idempotencyKey, ...payload } = job.data;
  const isNew = await deduplicate(`idempotency:${idempotencyKey}`);
  if (!isNew) {
    return { status: "already-processed" };
  }
  return await processOrder(payload);
});

Section 4: Testing Queue-Based Microservices

Testing async, queue-driven code requires a different approach than testing synchronous HTTP handlers.

Unit Testing Job Handlers

Extract the business logic from your job handlers into pure functions for easy unit testing:

// handlers/process-payment.ts (pure logic)
export async function processPayment(data: {
  orderId: string; amount: number; currency: string;
}): Promise<{ transactionId: string; status: string }> {
  // Business logic only — no queue dependency
  if (data.amount <= 0) throw new Error("Invalid amount");
  return { transactionId: `txn_${Date.now()}`, status: "success" };
}

// In your test:
import { describe, it, expect } from "vitest";
import { processPayment } from "./handlers/process-payment";

describe("processPayment", () => {
  it("processes a valid payment", async () => {
    const result = await processPayment({ orderId: "123", amount: 5000, currency: "usd" });
    expect(result.status).toBe("success");
    expect(result.transactionId).toMatch(/^txn_/);
  });

  it("rejects invalid amounts", async () => {
    await expect(processPayment({ orderId: "123", amount: -1, currency: "usd" }))
      .rejects.toThrow("Invalid amount");
  });
});

Integration Tests with Redis

For end-to-end tests that verify queue behavior, use Testcontainers to spin up a disposable Redis instance:

import { GenericContainer } from "testcontainers";
import Queue from "bee-queue";

describe("payment queue integration", () => {
  let redisContainer;
  let queue;

  beforeAll(async () => {
    redisContainer = await new GenericContainer("redis:7-alpine")
      .withExposedPorts(6379)
      .start();

    const redisUrl = `redis://${redisContainer.getHost()}:${redisContainer.getMappedPort(6379)}`;
    process.env.REDIS_URL = redisUrl;

    queue = new Queue("test-payments", { redis: redisUrl, isWorker: true });
  }, 30000);

  afterAll(async () => {
    await queue.close();
    await redisContainer.stop();
  });

  it("processes a job end-to-end", async () => {
    const result = await new Promise((resolve, reject) => {
      queue.process(async (job) => {
        return { processed: true, value: job.data.value * 2 };
      });

      const producer = new Queue("test-payments", { redis: process.env.REDIS_URL!, isWorker: false });
      const job = producer.createJob({ value: 21 });
      job.on("succeeded", (result) => resolve(result));
      job.on("failed", (err) => reject(err));
      job.save();
    });

    expect(result).toEqual({ processed: true, value: 42 });
  });
});

Section 5: Integration with Web Frameworks

Connecting web frameworks to BeeQueue requires thoughtful patterns to keep request handlers lean and error handling centralized.

Express Middleware

// middleware/enqueue-job.ts
import { Request, Response, NextFunction } from "express";
import Queue from "bee-queue";

export function enqueueJob(queue: Queue) {
  return async (req: Request, res: Response, next: NextFunction) => {
    try {
      const job = queue.createJob(req.body as Record<string, unknown>);
      job.on("succeeded", (result) => { res.json({ jobId: job.id, result }); });
      job.on("failed", (err) => { res.status(500).json({ error: err.message }); });
      await job.save();

      // If the client wants just the job ID (poll-based):
      // res.status(202).json({ jobId: job.id });
    } catch (err) {
      next(err);
    }
  };
}

// Usage:
app.post("/api/process-image", enqueueJob(imageProcessingQueue));

NestJS Module

For NestJS applications, a dedicated module provides clean dependency injection:

// beequeue.module.ts
import { Module, DynamicModule, Global } from "@nestjs/common";
import Queue from "bee-queue";

@Global()
@Module({})
export class BeeQueueModule {
  static forRoot(queues: Array<{ name: string; options?: ConstructorParameters<typeof Queue>[1] }>): DynamicModule {
    const providers = queues.map((q) => ({
      provide: `QUEUE_${q.name.toUpperCase()}`,
      useFactory: () => new Queue(q.name, q.options),
    }));

    return {
      module: BeeQueueModule,
      providers,
      exports: providers,
    };
  }
}

// In your app module:
@Module({
  imports: [
    BeeQueueModule.forRoot([
      { name: "email", options: { redis: process.env.REDIS_URL! } },
      { name: "payment", options: { redis: process.env.REDIS_URL! } },
    ]),
  ],
})
export class AppModule {}

Section 6: Custom Extensions and Advanced Patterns

Middleware Pipeline for Job Processing

Build a middleware layer for cross-cutting concerns like logging, validation, and metrics:

// lib/middleware.ts
type Middleware<T = unknown> = (
  job: import("bee-queue").Job<T>,
  next: () => Promise<unknown>,
) => Promise<unknown>;

export function applyMiddleware<T>(
  handler: (job: import("bee-queue").Job<T>) => Promise<unknown>,
  middlewares: Middleware<T>[],
) {
  return async (job: import("bee-queue").Job<T>) => {
    const composed = middlewares.reduceRight(
      (next, middleware) => async () => middleware(job, next),
      async () => handler(job),
    );
    return composed();
  };
}

// Example middleware:
const loggingMiddleware: Middleware = async (job, next) => {
  console.log(`[Job ${job.id}] Starting...`);
  const start = Date.now();
  try {
    const result = await next();
    console.log(`[Job ${job.id}] Completed in ${Date.now() - start}ms`);
    return result;
  } catch (err) {
    console.error(`[Job ${job.id}] Failed after ${Date.now() - start}ms:`, err);
    throw err;
  }
};

const validationMiddleware: Middleware = async (job, next) => {
  if (!job.data || typeof job.data !== "object") {
    throw new Error("Invalid job data: must be an object");
  }
  return next();
};

// Usage:
queue.process(applyMiddleware(myHandler, [loggingMiddleware, validationMiddleware]));

Job Router for Dynamic Handlers

Route jobs to different handlers based on a type field in the data:

// lib/job-router.ts
import Queue from "bee-queue";

type JobHandler<T = unknown> = (job: import("bee-queue").Job<T>) => Promise<unknown>;

export function createJobRouter<T extends { type: string }>() {
  const handlers = new Map<string, JobHandler<T>>();

  return {
    register(type: string, handler: JobHandler<T>) {
      handlers.set(type, handler);
    },
    getHandler(): (job: import("bee-queue").Job<T>) => Promise<unknown> {
      return async (job) => {
        const { type } = job.data;
        const handler = handlers.get(type);
        if (!handler) throw new Error(`No handler registered for job type: "${type}"`);
        return handler(job);
      };
    },
  };
}

// Usage:
const router = createJobRouter<{ type: string; payload: unknown }>();
router.register("send-email", async (job) => { /* send email */ });
router.register("generate-pdf", async (job) => { /* generate PDF */ });

queue.process(router.getHandler());

Production Readiness Checklist

Before deploying your BeeQueue-backed microservices to production, run through this checklist:

Graceful Shutdown:

SIGTERM handler stops processing new jobs and drains in-flight jobs
Kubernetes preStop hook with sufficient sleep time
terminationGracePeriodSeconds is longer than the shutdown timeout
Readiness probe returns 503 during shutdown

Error Handling:

Dead letter queue captures jobs that exhaust retries
Circuit breakers protect downstream service calls
Idempotency keys prevent duplicate processing
Failed job events are logged and monitored

Observability:

Queue health check endpoint (/healthz, /readyz)
Job processing metrics (throughput, latency, error rate)
Worker count and concurrency monitoring
Redis connection health monitoring

Reliability:

Reconnection strategy configured for Redis
Stalled job detection enabled (stallInterval)
At-least-once delivery understood and idempotency in place
Workload tested with simulated pod failures

Conclusion

BeeQueue may be small — roughly 1,000 lines of code — but it's a production-proven foundation for event-driven microservices. Its simplicity is its strength: you can go from zero to a functioning saga pattern, graceful shutdown, dead letter queue, and circuit breaker without leaving the Redis ecosystem you probably already run.

The patterns in this guide — request/reply, pub/sub event bus, sagas, graceful shutdown in Kubernetes, dead letter queues, circuit breakers, testing strategies, and framework integration — give you a production-ready architecture that scales with your services.

When you're running BeeQueue in production, you need visibility into your queues across all services and environments. QueueHub supports BeeQueue queues natively, giving you real-time dashboards, job management, live worker views, and multi-environment monitoring — all without the operational overhead of self-hosted solutions.

Try QueueHub for free and get instant visibility into your BeeQueue microservices architecture.