·QueueHub Team·24 min read

Real-Time Queue Dashboards Without Polling: Event-Driven Architecture with BullMQ QueueEvents

BullMQ QueueEventsreal-time dashboardWebSocketSSEevent-driven monitoringqueue observabilityalerting

Introduction — Why Polling Is the Wrong Default

Most queue dashboards follow the same pattern: every N seconds, the browser fires off a batch of API calls — /jobs?state=completed, /counts, /metrics — and the server dutifully hits Redis for every request. This works when you're watching a single queue with a handful of jobs per minute. But under production load, the cracks appear: redundant queries that return the same data, stale windows between polls where failures go unnoticed, and Redis CPU spikes as every client competes for the same keys.

There's a better way. BullMQ's QueueEvents class uses Redis Streams to deliver every job lifecycle event — completed, failed, active, progress, stalled, delayed, drained — to any subscriber, in any process, in real time. Your dashboard becomes a subscriber rather than a poller.

In this post, you'll learn:

  • How BullMQ QueueEvents works under the hood with Redis Streams
  • Streaming events to browsers via Server-Sent Events (SSE) and WebSockets
  • Building a real-time metrics aggregator for throughput, error rates, and P50/P95/P99 latency
  • Triggering alerting pipelines (Slack, email, PagerDuty) directly from queue events
  • Multi-tenant event filtering so each team sees only their own jobs
  • Production considerations for connection management, backpressure, and horizontal scaling

Prerequisites: Node.js 18+, BullMQ v5+, and basic familiarity with Queue, Worker, and Job.

BullMQ QueueEvents — The Backbone of Event-Driven Monitoring

How QueueEvents Works Under the Hood

Worker-level event listeners (worker.on('completed', ...)) only fire in the same process where the worker is running. If your dashboard server is a separate process — which it should be — those events never reach it.

QueueEvents solves this by acting as a dedicated consumer that reads from Redis Streams. Under the hood, BullMQ writes every job lifecycle event to a Redis stream keyed as bull:{queueName}:events. A QueueEvents instance uses XREADGROUP to consume from that stream, delivering events in order with at-least-once semantics.

Key properties:

  • Cross-process: A separate Node.js process, or even a different runtime, can consume events from queues it never created
  • Durable: Redis Streams persist events until trimmed. A temporary disconnection doesn't lose events — the consumer resumes from where it left off
  • Ordered: Events within each stream shard arrive in the order they were produced
  • Configurable retention: Default ~10,000 events per queue, adjustable via streams.events.maxLen
import { QueueEvents } from 'bullmq';
import { Redis } from 'ioredis';

const connection = new Redis({
  host: 'localhost',
  port: 6379,
  maxRetriesPerRequest: null,
});

const queueEvents = new QueueEvents('pipeline-jobs', {
  connection,
  // Keep last 50,000 events in the stream
  streams: { events: { maxLen: 50000 } },
});

All Available QueueEvents — Complete Reference

BullMQ fires a distinct event for every job lifecycle transition. Here is the full set available in BullMQ v5:

interface QueueEventsListener {
  'active':      ({ jobId, prev }: { jobId: string; prev: string }) => void;
  'added':       ({ jobId, name, data }: { jobId: string; name: string; data: string }) => void;
  'cleaned':     ({ count }: { count: number }) => void;
  'completed':   ({ jobId, returnvalue }: { jobId: string; returnvalue: string }) => void;
  'delayed':     ({ jobId, delay }: { jobId: string; delay: number }) => void;
  'drained':     () => void;
  'failed':      ({ jobId, failedReason }: { jobId: string; failedReason: string }) => void;
  'paused':      () => void;
  'progress':    ({ jobId, data }: { jobId: string; data: number | object }) => void;
  'removed':     ({ jobId }: { jobId: string }) => void;
  'resumed':     () => void;
  'stalled':     ({ jobId }: { jobId: string }) => void;
  'waiting':     ({ jobId }: { jobId: string }) => void;
  'waiting-children': ({ jobId }: { jobId: string }) => void;
}

queueEvents.on('waiting', ({ jobId }) => {
  console.log(`Job ${jobId} entered the waiting queue`);
});

queueEvents.on('progress', ({ jobId, data }) => {
  console.log(`Job ${jobId} reported progress:`, data);
});

queueEvents.on('failed', ({ jobId, failedReason }) => {
  console.error(`Job ${jobId} failed:`, failedReason);
});

Lifecycle-Aware State Tracking

A single JobStateTracker class can maintain an in-memory map of every job's current state by subscribing to all events. This gives your dashboard a live, queryable view of every job without ever calling queue.getJob():

type JobState = 'waiting' | 'waiting-children' | 'active' | 'delayed'
  | 'completed' | 'failed' | 'stalled' | 'removed';

interface TrackedJob {
  id: string;
  name: string;
  state: JobState;
  progress: number | object | null;
  data: unknown;
  result: unknown;
  error: string | null;
  timestamps: {
    added: number | null;
    started: number | null;
    completed: number | null;
    failed: number | null;
    delayed: number | null;
  };
  processingDuration: number | null;
}

class JobStateTracker {
  private jobs = new Map<string, TrackedJob>();
  private queueEvents: QueueEvents;

  constructor(queueName: string, connection: Redis) {
    this.queueEvents = new QueueEvents(queueName, { connection });
    this.registerListeners();
  }

  private registerListeners(): void {
    this.queueEvents.on('waiting', ({ jobId }) => {
      this.updateJob(jobId, { state: 'waiting', timestamps: { ...this.getOrCreate(jobId).timestamps, added: Date.now() } });
    });

    this.queueEvents.on('active', ({ jobId }) => {
      this.updateJob(jobId, { state: 'active', timestamps: { ...this.getOrCreate(jobId).timestamps, started: Date.now() } });
    });

    this.queueEvents.on('progress', ({ jobId, data }) => {
      this.updateJob(jobId, { progress: data });
    });

    this.queueEvents.on('completed', ({ jobId, returnvalue }) => {
      const job = this.jobs.get(jobId);
      const now = Date.now();
      this.updateJob(jobId, {
        state: 'completed',
        result: returnvalue,
        timestamps: { ...job?.timestamps, completed: now } as TrackedJob['timestamps'],
        processingDuration: job?.timestamps.started ? now - job.timestamps.started : null,
      });
    });

    this.queueEvents.on('failed', ({ jobId, failedReason }) => {
      const job = this.jobs.get(jobId);
      const now = Date.now();
      this.updateJob(jobId, {
        state: 'failed',
        error: failedReason,
        timestamps: { ...job?.timestamps, failed: now } as TrackedJob['timestamps'],
        processingDuration: job?.timestamps.started ? now - job.timestamps.started : null,
      });
    });

    this.queueEvents.on('stalled', ({ jobId }) => {
      this.updateJob(jobId, { state: 'stalled' });
    });

    this.queueEvents.on('removed', ({ jobId }) => {
      this.jobs.delete(jobId);
    });
  }

  private getOrCreate(jobId: string): TrackedJob {
    const existing = this.jobs.get(jobId);
    if (existing) return existing;
    const created: TrackedJob = {
      id: jobId, name: '', state: 'waiting', progress: null,
      data: null, result: null, error: null,
      timestamps: { added: null, started: null, completed: null, failed: null, delayed: null },
      processingDuration: null,
    };
    this.jobs.set(jobId, created);
    return created;
  }

  private updateJob(jobId: string, partial: Partial<TrackedJob>): void {
    const existing = this.jobs.get(jobId);
    if (existing) {
      Object.assign(existing, partial);
    } else {
      this.jobs.set(jobId, { ...partial as any, id: jobId });
    }
  }

  getJob(jobId: string): TrackedJob | undefined {
    return this.jobs.get(jobId);
  }

  getAllJobs(): TrackedJob[] {
    return Array.from(this.jobs.values());
  }

  async close(): Promise<void> {
    await this.queueEvents.close();
  }
}

This tracker can feed directly into your dashboard's state — no polling, no redundant Redis queries, just events flowing in and state flowing out.

Pushing Events to the Browser — WebSocket vs. SSE

Choosing the Right Transport

Once QueueEvents are delivering job lifecycle data to your server, you need a way to push that data to browser clients in real time. You have two main options:

Criterion WebSocket (ws / Socket.IO) SSE (EventSource)
Direction Bidirectional Server to client only
Reconnection Built-in (Socket.IO) Built-in (browser EventSource)
Binary support Yes No (text only)
HTTP/2 multiplexing No (separate connection) Yes (shared connection)
Browser support Universal Universal (except IE)
Backpressure handling Manual Automatic (TCP backpressure)
Complexity Higher Lower

Recommendation: Use SSE for broadcast-style dashboards where one queue serves many viewers and the client only needs to receive data. Use WebSockets when clients need to send filter commands, subscribe/unsubscribe to specific queues, or interact with the dashboard in real time.

SSE Implementation — QueueEvents to Browser Stream

SSE is the simpler option and works well for most queue dashboards. Here's how to wire up an Express endpoint that streams BullMQ events directly to the browser:

import { Router, Request, Response } from 'express';
import { QueueEvents } from 'bullmq';
import { Redis } from 'ioredis';

const SSE_HEADERS = {
  'Content-Type': 'text/event-stream',
  'Cache-Control': 'no-cache',
  'Connection': 'keep-alive',
  'X-Accel-Buffering': 'no',
};

function createSSERouter(connection: Redis): Router {
  const router = Router();

  router.get('/api/queues/:queueName/stream', (req: Request, res: Response) => {
    const { queueName } = req.params;
    res.writeHead(200, SSE_HEADERS);

    const queueEvents = new QueueEvents(queueName, { connection });
    let eventId = 0;

    // Send an initial keepalive
    res.write(':connected\n\n');

    const heartbeat = setInterval(() => {
      res.write(':heartbeat\n\n');
    }, 15000);

    // Map every QueueEvents type to an SSE event
    const emitSSE = (event: string, data: unknown) => {
      eventId++;
      res.write(`id: ${eventId}\nevent: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
    };

    queueEvents.on('waiting',    ({ jobId }) => emitSSE('waiting', { jobId }));
    queueEvents.on('active',     ({ jobId }) => emitSSE('active', { jobId }));
    queueEvents.on('progress',   ({ jobId, data }) => emitSSE('progress', { jobId, data }));
    queueEvents.on('completed',  ({ jobId, returnvalue }) => emitSSE('completed', { jobId, returnvalue }));
    queueEvents.on('failed',     ({ jobId, failedReason }) => emitSSE('failed', { jobId, failedReason }));
    queueEvents.on('stalled',    ({ jobId }) => emitSSE('stalled', { jobId }));
    queueEvents.on('delayed',    ({ jobId }) => emitSSE('delayed', { jobId }));
    queueEvents.on('drained',    () => emitSSE('drained', {}));

    req.on('close', async () => {
      clearInterval(heartbeat);
      await queueEvents.close();
    });
  });

  return router;
}

On the client side, consuming this stream is trivial with the native EventSource API:

function connectQueueStream(queueName: string): EventSource {
  const source = new EventSource(`/api/queues/${queueName}/stream`);

  source.addEventListener('completed', (event: MessageEvent) => {
    const { jobId, returnvalue } = JSON.parse(event.data);
    updateDashboard({ type: 'completed', jobId, returnvalue });
  });

  source.addEventListener('failed', (event: MessageEvent) => {
    const { jobId, failedReason } = JSON.parse(event.data);
    updateDashboard({ type: 'failed', jobId, failedReason });
  });

  source.addEventListener('progress', (event: MessageEvent) => {
    const { jobId, data } = JSON.parse(event.data);
    updateDashboard({ type: 'progress', jobId, progress: data });
  });

  source.onerror = () => {
    console.warn('SSE connection lost, reconnecting...');
    // EventSource auto-reconnects
  };

  return source;
}

No polling, no manual refresh button — the browser receives events as they happen.

WebSocket Implementation — Socket.IO with Room-Based Subscriptions

When clients need to dynamically choose which queues to watch, Socket.IO provides room-based subscriptions that make multi-queue dashboards straightforward:

import { Server as SocketIOServer } from 'socket.io';
import { Server as HTTPServer } from 'http';
import { QueueEvents } from 'bullmq';
import { Redis } from 'ioredis';

interface QueueSubscription {
  queueEvents: QueueEvents;
  queueName: string;
  roomId: string;
}

export function createSocketIOBackend(httpServer: HTTPServer, connection: Redis): SocketIOServer {
  const io = new SocketIOServer(httpServer, {
    cors: { origin: '*' },
    transports: ['websocket', 'polling'],
  });

  const activeSubscriptions = new Map<string, QueueSubscription>();

  io.on('connection', (socket) => {
    socket.on('subscribe:queue', async (queueName: string) => {
      const roomId = `queue:${queueName}`;
      socket.join(roomId);

      // Only create one QueueEvents listener per queue regardless of client count
      if (!activeSubscriptions.has(roomId)) {
        const queueEvents = new QueueEvents(queueName, { connection });

        queueEvents.on('completed', ({ jobId, returnvalue }) => {
          io.to(roomId).emit('job:completed', { jobId, queueName, returnvalue });
        });

        queueEvents.on('failed', ({ jobId, failedReason }) => {
          io.to(roomId).emit('job:failed', { jobId, queueName, failedReason });
        });

        queueEvents.on('progress', ({ jobId, data }) => {
          io.to(roomId).emit('job:progress', { jobId, queueName, progress: data });
        });

        queueEvents.on('active', ({ jobId, prev }) => {
          io.to(roomId).emit('job:active', { jobId, queueName, prev });
        });

        queueEvents.on('waiting', ({ jobId }) => {
          io.to(roomId).emit('job:waiting', { jobId, queueName });
        });

        queueEvents.on('stalled', ({ jobId }) => {
          io.to(roomId).emit('job:stalled', { jobId, queueName });
        });

        queueEvents.on('delayed', ({ jobId, delay }) => {
          io.to(roomId).emit('job:delayed', { jobId, queueName, delay });
        });

        queueEvents.on('drained', () => {
          io.to(roomId).emit('queue:drained', { queueName });
        });

        activeSubscriptions.set(roomId, { queueEvents, queueName, roomId });
      }

      socket.emit('subscribed', { queueName });
    });

    socket.on('unsubscribe:queue', (queueName: string) => {
      const roomId = `queue:${queueName}`;
      socket.leave(roomId);
    });
  });

  return io;
}

The key insight in this implementation: we create one QueueEvents instance per queue, regardless of how many clients are watching it. The activeSubscriptions map ensures we never duplicate the Redis Stream consumer, and Socket.IO rooms handle fan-out to all connected clients efficiently.

Custom Metric Aggregators — Turning Events Into Real-Time Analytics

Raw job events are useful, but what your dashboard really needs is derived metrics: throughput rates, error percentages, and latency percentiles that update in real time. The aggregator pattern lets you compute these from the event stream without touching Redis.

Sliding-Window Throughput Counter

This aggregator tracks completed and failed jobs per minute over a configurable window:

interface ThroughputBucket {
  timestamp: number; // epoch seconds (floor to minute)
  completed: number;
  failed: number;
}

class ThroughputAggregator {
  private windowSizeMinutes: number;
  private buckets: ThroughputBucket[] = [];

  constructor(windowSizeMinutes: number = 60) {
    this.windowSizeMinutes = windowSizeMinutes;
  }

  recordCompleted(): void {
    this.touchBucket().completed++;
  }

  recordFailed(): void {
    this.touchBucket().failed++;
  }

  private touchBucket(): ThroughputBucket {
    const now = Math.floor(Date.now() / 60000) * 60;
    let bucket = this.buckets.find((b) => b.timestamp === now);
    if (!bucket) {
      bucket = { timestamp: now, completed: 0, failed: 0 };
      this.buckets.push(bucket);
      this.evictOld();
    }
    return bucket;
  }

  private evictOld(): void {
    const cutoff = Math.floor(Date.now() / 1000) - this.windowSizeMinutes * 60;
    this.buckets = this.buckets.filter((b) => b.timestamp >= cutoff);
  }

  getThroughputRates(): { completedPerMin: number; failedPerMin: number } {
    if (this.buckets.length === 0) return { completedPerMin: 0, failedPerMin: 0 };
    const totalCompleted = this.buckets.reduce((s, b) => s + b.completed, 0);
    const totalFailed = this.buckets.reduce((s, b) => s + b.failed, 0);
    const minutes = this.buckets.length;
    return {
      completedPerMin: Math.round((totalCompleted / minutes) * 100) / 100,
      failedPerMin: Math.round((totalFailed / minutes) * 100) / 100,
    };
  }

  getTimeSeries(): ThroughputBucket[] {
    return [...this.buckets];
  }
}

The sliding-window approach automatically evicts old data — no background cleanup job required. The getTimeSeries() method can feed a chart component that updates as new buckets fill.

Running Error Rate Calculator

Error rate is one of the most important operational metrics. This aggregator maintains a sliding window of the most recent jobs:

class ErrorRateAggregator {
  private total = 0;
  private errors = 0;
  private windowJobs: Array<{ succeeded: boolean; timestamp: number }> = [];
  private windowMs: number;

  constructor(windowMs: number = 300_000) {
    this.windowMs = windowMs;
  }

  recordCompletion(): void {
    this.total++;
    this.windowJobs.push({ succeeded: true, timestamp: Date.now() });
    this.evict();
  }

  recordFailure(): void {
    this.total++;
    this.errors++;
    this.windowJobs.push({ succeeded: false, timestamp: Date.now() });
    this.evict();
  }

  private evict(): void {
    const cutoff = Date.now() - this.windowMs;
    while (this.windowJobs.length > 0 && this.windowJobs[0].timestamp < cutoff) {
      const evicted = this.windowJobs.shift()!;
      if (!evicted.succeeded) this.errors--;
      this.total--;
    }
  }

  getErrorRate(): number {
    return this.total > 0 ? this.errors / this.total : 0;
  }

  getTotalProcessed(): number {
    return this.total;
  }

  getTotalErrors(): number {
    return this.errors;
  }
}

P50 / P95 / P99 Latency Tracker

Processing latency percentiles are the gold standard for queue health. To compute them from events, we pair active events (which record when processing started) with completed events (which record when it finished):

class LatencyPercentileTracker {
  private activeTimestamps = new Map<string, number>();
  private durations: number[] = [];
  private maxSamples: number;

  constructor(maxSamples: number = 10_000) {
    this.maxSamples = maxSamples;
  }

  recordActive(jobId: string): void {
    this.activeTimestamps.set(jobId, Date.now());
  }

  recordCompleted(jobId: string): void {
    const started = this.activeTimestamps.get(jobId);
    if (started === undefined) return;
    const duration = Date.now() - started;
    this.durations.push(duration);
    this.activeTimestamps.delete(jobId);

    // Keep the window bounded
    if (this.durations.length > this.maxSamples) {
      this.durations.shift();
    }
  }

  private percentile(p: number): number | null {
    if (this.durations.length === 0) return null;
    const sorted = [...this.durations].sort((a, b) => a - b);
    const index = Math.ceil((p / 100) * sorted.length) - 1;
    return sorted[Math.max(0, index)];
  }

  getP50(): number | null { return this.percentile(50); }
  getP95(): number | null { return this.percentile(95); }
  getP99(): number | null { return this.percentile(99); }

  getSlate(): { p50: number | null; p95: number | null; p99: number | null; sampleCount: number } {
    return {
      p50: this.getP50(),
      p95: this.getP95(),
      p99: this.getP99(),
      sampleCount: this.durations.length,
    };
  }
}

Composing the Aggregators Into a Single Event Consumer

These three aggregators work best when composed into a single coordinator that maps QueueEvents to the right aggregator calls:

interface MetricsSnapshot {
  throughput: { completedPerMin: number; failedPerMin: number };
  errorRate: number;
  latency: { p50: number | null; p95: number | null; p99: number | null; sampleCount: number };
  timestamp: number;
}

class RealTimeMetricsCollector {
  readonly throughput: ThroughputAggregator;
  readonly errorRate: ErrorRateAggregator;
  readonly latency: LatencyPercentileTracker;

  constructor(windowMinutes: number = 60) {
    this.throughput = new ThroughputAggregator(windowMinutes);
    this.errorRate = new ErrorRateAggregator();
    this.latency = new LatencyPercentileTracker(10_000);
  }

  handleActive({ jobId }: { jobId: string }): void {
    this.latency.recordActive(jobId);
  }

  handleCompleted({ jobId }: { jobId: string }): void {
    this.throughput.recordCompleted();
    this.errorRate.recordCompletion();
    this.latency.recordCompleted(jobId);
  }

  handleFailed({ jobId }: { jobId: string }): void {
    this.throughput.recordFailed();
    this.errorRate.recordFailure();
    this.latency.recordCompleted(jobId);
  }

  getSnapshot(): MetricsSnapshot {
    return {
      throughput: this.throughput.getThroughputRates(),
      errorRate: this.errorRate.getErrorRate(),
      latency: this.latency.getSlate(),
      timestamp: Date.now(),
    };
  }
}

Wire this collector into the SSE endpoint to emit a metrics event every 5 seconds — the browser gets a steady stream of precomputed analytics without any polling:

// Inside the SSE endpoint handler
const interval = setInterval(() => {
  res.write(`event: metrics\ndata: ${JSON.stringify(metrics.getSnapshot())}\n\n`);
}, 5000);

SLA Tracking and Trend Analysis

Real-time metrics are most valuable when compared against service level agreements (SLAs). An SLA rule is a simple threshold expression evaluated against the current metrics snapshot.

Defining SLA Rules

interface SLARule {
  name: string;
  metric: 'p95' | 'p99' | 'errorRate' | 'throughput';
  operator: 'lt' | 'gt' | 'lte' | 'gte';
  threshold: number;
  severity: 'warning' | 'critical';
}

class SLAMonitor {
  private rules: SLARule[] = [];
  private violations: Array<{ rule: SLARule; value: number; timestamp: number }> = [];

  constructor(rules: SLARule[]) {
    this.rules = rules;
  }

  evaluate(snapshot: MetricsSnapshot): Array<{ ruleName: string; current: number; threshold: number; severity: string }> {
    const breaches: Array<{ ruleName: string; current: number; threshold: number; severity: string }> = [];

    for (const rule of this.rules) {
      let value: number | null = null;

      switch (rule.metric) {
        case 'p95': value = snapshot.latency.p95; break;
        case 'p99': value = snapshot.latency.p99; break;
        case 'errorRate': value = snapshot.errorRate; break;
        case 'throughput': value = snapshot.throughput.completedPerMin; break;
      }

      if (value === null) continue;

      const breached = this.compare(value, rule.threshold, rule.operator);
      if (breached) {
        const violation = { rule, value, timestamp: Date.now() };
        this.violations.push(violation);
        breaches.push({ ruleName: rule.name, current: value, threshold: rule.threshold, severity: rule.severity });
      }
    }

    return breaches;
  }

  private compare(value: number, threshold: number, op: string): boolean {
    switch (op) {
      case 'lt':  return value < threshold;
      case 'gt':  return value > threshold;
      case 'lte': return value <= threshold;
      case 'gte': return value >= threshold;
      default:    return false;
    }
  }

  getViolationHistory(): typeof this.violations {
    return [...this.violations];
  }
}

Example SLA rules for a production queue:

const slaMonitor = new SLAMonitor([
  { name: 'P95 under 500ms', metric: 'p95', operator: 'lt', threshold: 500, severity: 'warning' },
  { name: 'Error rate under 1%', metric: 'errorRate', operator: 'lt', threshold: 0.01, severity: 'critical' },
  { name: 'Throughput above 10/min', metric: 'throughput', operator: 'gte', threshold: 10, severity: 'warning' },
]);

On the dashboard, SLA violations can drive:

  • A health badge per queue (green / yellow / red) that flips immediately when a threshold is crossed
  • A trend line showing P95 over the last 60 minutes
  • An SLA calendar showing how many minutes each queue was in breach over the current hour

Persisting Metrics for Historical Queries

The in-memory aggregators only cover the sliding window (last 60 minutes by default). For queries spanning days or weeks, persist snapshot data to Redis or a time-series database:

async function persistMetrics(
  redis: Redis,
  queueName: string,
  snapshot: MetricsSnapshot
): Promise<void> {
  const key = `metrics:${queueName}:${Math.floor(Date.now() / 60000)}`;
  await redis.set(key, JSON.stringify(snapshot), 'EX', 86400 * 30); // 30-day TTL
}

// Call this every 60 seconds from a setInterval
setInterval(async () => {
  for (const [queueName, collector] of metricsMap.entries()) {
    await persistMetrics(redis, queueName, collector.getSnapshot());
  }
}, 60000);

Alerting Integrations — From Queue Event to Notification

Perhaps the most powerful pattern: triggering alerts (Slack, email, PagerDuty) directly from QueueEvents, with no polling or separate alert-checking daemon.

Alert Rule Definition

interface AlertRule {
  name: string;
  eventType: 'completed' | 'failed' | 'stalled' | 'delayed' | 'progress' | 'drained';
  condition: 'always' | 'threshold';
  threshold?: {
    field: 'failedReason' | 'returnvalue' | 'delay' | 'progress';
    operator: 'contains' | 'equals' | 'gt' | 'lt';
    value: string | number;
  };
  channels: AlertChannel[];
  cooldownMs: number;
  lastFiredAt: number;
}

type AlertChannel = SlackWebhookChannel | EmailChannel | PagerDutyChannel;

interface SlackWebhookChannel {
  type: 'slack';
  webhookUrl: string;
  messageTemplate: string;
}

interface EmailChannel {
  type: 'email';
  to: string[];
  subjectTemplate: string;
  bodyTemplate: string;
}

interface PagerDutyChannel {
  type: 'pagerduty';
  routingKey: string;
  severity: 'info' | 'warning' | 'error' | 'critical';
  dedupKeyTemplate: string;
}

Alert Engine

The engine evaluates rules against events and dispatches through the configured channels:

class AlertEngine {
  private rules: AlertRule[] = [];

  addRule(rule: AlertRule): void {
    this.rules.push(rule);
  }

  async evaluate(eventType: string, eventPayload: Record<string, unknown>, queueName: string): Promise<void> {
    const now = Date.now();

    for (const rule of this.rules) {
      if (rule.eventType !== eventType) continue;
      if (now - rule.lastFiredAt < rule.cooldownMs) continue;

      if (rule.condition === 'always' || this.matchesThreshold(rule.threshold!, eventPayload)) {
        rule.lastFiredAt = now;

        await Promise.all(
          rule.channels.map((channel) => this.dispatch(channel, { ...eventPayload, queueName, ruleName: rule.name }))
        );
      }
    }
  }

  private matchesThreshold(threshold: AlertRule['threshold'], payload: Record<string, unknown>): boolean {
    if (!threshold) return true;
    const value = payload[threshold.field];
    if (value === undefined) return false;

    switch (threshold.operator) {
      case 'contains': return String(value).includes(String(threshold.value));
      case 'equals':   return value === threshold.value;
      case 'gt':       return Number(value) > Number(threshold.value);
      case 'lt':       return Number(value) < Number(threshold.value);
      default:         return false;
    }
  }

  private async dispatch(channel: AlertChannel, context: Record<string, unknown>): Promise<void> {
    switch (channel.type) {
      case 'slack':
        return this.sendSlackAlert(channel as SlackWebhookChannel, context);
      case 'email':
        return this.sendEmailAlert(channel as EmailChannel, context);
      case 'pagerduty':
        return this.sendPagerDutyAlert(channel as PagerDutyChannel, context);
    }
  }

  private async sendSlackAlert(channel: SlackWebhookChannel, context: Record<string, unknown>): Promise<void> {
    const message = this.interpolate(channel.messageTemplate, context);
    await fetch(channel.webhookUrl, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ text: message }),
    });
  }

  private async sendEmailAlert(channel: EmailChannel, context: Record<string, unknown>): Promise<void> {
    const subject = this.interpolate(channel.subjectTemplate, context);
    const body = this.interpolate(channel.bodyTemplate, context);
    // Integrate with nodemailer or SendGrid here
    console.log(`[Alert Email] To: ${channel.to.join(',')} Subject: ${subject}`);
  }

  private async sendPagerDutyAlert(channel: PagerDutyChannel, context: Record<string, unknown>): Promise<void> {
    const dedupKey = this.interpolate(channel.dedupKeyTemplate, context);
    await fetch('https://events.pagerduty.com/v2/enqueue', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        routing_key: channel.routingKey,
        event_action: 'trigger',
        dedup_key: dedupKey,
        payload: {
          summary: context.failedReason
            ? `Job failed: ${context.failedReason}`
            : `Queue event: ${context.eventType}`,
          source: context.queueName as string,
          severity: channel.severity,
          custom_details: context,
        },
      }),
    });
  }

  private interpolate(template: string, context: Record<string, unknown>): string {
    return template.replace(/\{(\w+)\}/g, (_, key) => String(context[key] ?? ''));
  }
}

Wiring Alerts Into the Event Consumer

const alertEngine = new AlertEngine();

alertEngine.addRule({
  name: 'Critical job failure',
  eventType: 'failed',
  condition: 'threshold',
  threshold: { field: 'failedReason', operator: 'contains', value: 'OOM' },
  channels: [
    {
      type: 'slack',
      webhookUrl: process.env.SLACK_WEBHOOK_URL!,
      messageTemplate: ':fire: *CRITICAL* — Job `{jobId}` on `{queueName}` failed with OOM: `{failedReason}`',
    },
    {
      type: 'pagerduty',
      routingKey: process.env.PAGERDUTY_ROUTING_KEY!,
      severity: 'critical',
      dedupKeyTemplate: 'bullmq-{queueName}-{jobId}',
    },
  ],
  cooldownMs: 60_000,
  lastFiredAt: 0,
});

alertEngine.addRule({
  name: 'Any job failure notification',
  eventType: 'failed',
  condition: 'always',
  channels: [
    {
      type: 'slack',
      webhookUrl: process.env.SLACK_WEBHOOK_URL!,
      messageTemplate: 'Job `{jobId}` failed on `{queueName}`: `{failedReason}`',
    },
  ],
  cooldownMs: 30_000,
  lastFiredAt: 0,
});

// Hook into the event consumer
queueEvents.on('failed', (payload) => {
  alertEngine.evaluate('failed', payload as unknown as Record<string, unknown>, 'my-queue');
});

Multi-Tenant Isolation — Filtering Events by Org/Team

In a multi-org SaaS like Queue Hub, a single Redis instance hosts queues for dozens of organizations. Raw QueueEvents streams emit every job lifecycle event globally — you must filter at the application layer before pushing to the browser.

Tenant Key Patterns

Pattern 1 — Prefix-based queue naming: Name queues with a tenant prefix so the tenant is extractable from the queue name alone:

queues:
  acme:email-worker
  acme:payment-processor
  bigco:email-worker
  bigco:image-resize

The SSE or WebSocket endpoint then only serves events for queues that belong to the authenticated organization. No per-event metadata lookup required.

Pattern 2 — Job metadata tagging: Even within a shared queue, jobs carry a tenantId in their data payload. The event consumer fetches the job metadata before forwarding:

queueEvents.on('completed', async ({ jobId, returnvalue }) => {
  const job = await queue.getJob(jobId);
  const tenantId = job?.data.tenantId;

  if (!tenantId) return;

  // Only broadcast to sockets subscribed to this tenant
  io.to(`tenant:${tenantId}`).emit('job:completed', {
    jobId, queueName: 'shared-queue', returnvalue
  });
});

Tenant Filter Middleware for SSE

router.get('/api/orgs/:orgId/queues/:queueName/stream', (req: Request, res: Response) => {
  const { orgId, queueName } = req.params;

  // Verify the org owns this queue (database lookup or token claim)
  const canAccess = verifyOrgQueueAccess(orgId, queueName);
  if (!canAccess) {
    res.status(403).json({ error: 'Forbidden' });
    return;
  }

  res.writeHead(200, SSE_HEADERS);

  const queueEvents = new QueueEvents(queueName, { connection });

  const emitIfOwned = (event: string, payload: Record<string, unknown>) => {
    // Queue naming already enforces isolation, so no per-event check needed
    res.write(`event: ${event}\ndata: ${JSON.stringify(payload)}\n\n`);
  };

  queueEvents.on('completed', ({ jobId, returnvalue }) => emitIfOwned('completed', { jobId, returnvalue }));
  queueEvents.on('failed', ({ jobId, failedReason }) => emitIfOwned('failed', { jobId, failedReason }));
  queueEvents.on('active', ({ jobId, prev }) => emitIfOwned('active', { jobId, prev }));
  queueEvents.on('progress', ({ jobId, data }) => emitIfOwned('progress', { jobId, data }));

  req.on('close', () => queueEvents.close());
});

Org-Level Metrics Aggregation

The RealTimeMetricsCollector can be instantiated per org, giving each organization its own throughput, error rate, and latency calculations:

const collectors = new Map<string, RealTimeMetricsCollector>();

function getCollector(orgId: string): RealTimeMetricsCollector {
  if (!collectors.has(orgId)) {
    collectors.set(orgId, new RealTimeMetricsCollector(60));
  }
  return collectors.get(orgId)!;
}

queueEvents.on('completed', async ({ jobId }) => {
  const job = await queue.getJob(jobId);
  const orgId = job?.data.orgId ?? 'unknown';
  getCollector(orgId).handleCompleted({ jobId });
});

Production Considerations

Connection Management

  • One QueueEvents instance per queue, not per client. Reuse the same instance across all SSE/WebSocket connections for that queue.
  • Graceful shutdown: Always call queueEvents.close() on server shutdown and connection close to avoid leaving orphaned consumers in Redis.
  • Separate Redis connections: Use a dedicated Redis connection for QueueEvents to avoid blocking other operations:
const eventsRedis = new Redis({
  host: process.env.REDIS_HOST,
  port: Number(process.env.REDIS_PORT),
  maxRetriesPerRequest: null,
  enableReadyCheck: false,
});

Handling Backpressure

When event volume exceeds client consumption speed:

  • SSE naturally applies TCP backpressure — res.write() will buffer and eventually block
  • Use a ring buffer between QueueEvents and the transport to drop old events when the client is slow
  • Set maxListeners to avoid memory leaks: queueEvents.setMaxListeners(100)

Event Stream Trimming

BullMQ trims the events stream by default to ~10,000 events, but under high throughput this may fill up quickly:

setInterval(async () => {
  const queue = new Queue(queueName, { connection });
  await queue.trimEvents(10_000); // Keep 10k most recent events
}, 60_000);

Scaling Across Processes

For horizontally scaled dashboard servers, use Redis Pub/Sub as a middle layer to fan out events from a single QueueEvents consumer to multiple dashboard server instances:

// In the listener process: consumes QueueEvents, publishes to Redis Pub/Sub
const pub = new Redis({ host: 'redis', maxRetriesPerRequest: null });
queueEvents.on('completed', (payload) => {
  pub.publish('queue:dashboard', JSON.stringify({
    event: 'completed',
    queueName: 'email-worker',
    payload,
  }));
});

// In each dashboard server process: subscribes to Pub/Sub, pushes to local clients
const sub = new Redis({ host: 'redis', maxRetriesPerRequest: null });
sub.subscribe('queue:dashboard');
sub.on('message', (channel, message) => {
  const { event, queueName, payload } = JSON.parse(message);
  broadcastToLocalClients(event, queueName, payload);
});

Comparison: Event-Driven vs. Polling Dashboard

Dimension Polling Dashboard Event-Driven Dashboard
Redis load N queries per N seconds per client 1 stream per queue, independent of client count
Data freshness Up to N seconds stale Sub-millisecond
Scalability Degrades with client count Scales with queue count, not client count
Complexity Simpler to implement Requires event pipeline + transport
Resilience Blind spot between polls Events are durable in Redis Streams
Ideal for Low-traffic queues, admin UIs High-throughput queues, real-time ops

Conclusion

Moving from polling to event-driven architecture transforms your queue dashboard from a periodically refreshed page into a live operations center. By leveraging BullMQ's QueueEvents with Redis Streams, you eliminate redundant Redis queries, reduce latency from seconds to milliseconds, and unlock patterns — real-time metrics aggregation, instant SLA breach detection, and event-driven alerting — that are impractical with polling.

The pattern is straightforward: QueueEvents delivers the raw events, aggregators derive the metrics, SSE or WebSockets push them to the browser, and the alert engine triggers notifications when things go wrong. Each piece is independently testable, replaceable, and scalable.

Ready to take your queue monitoring to the next level? Queue Hub provides a production-ready dashboard with built-in event-driven monitoring, multi-tenant isolation, and alerting integrations — so you can skip the infrastructure and start observing your queues in real time today.

← All Articles
Share on XShare on LinkedIn

Related Articles

2026-07-01

Building a Queue Dashboard: Event-Driven Real-Time Monitoring with BullMQ

Stop polling Redis. Learn how to build a real-time BullMQ dashboard using BullMQ's event-driven architecture — QueueEvents, WebSocket push, job lifecycle visualization, and sliding-window metrics aggregation.

2026-06-30

Queue Monitoring Best Practices: A Complete Guide for BullMQ

Learn essential queue monitoring best practices for BullMQ — metrics, dashboards, alerting, and observability patterns to keep your job queues healthy and performant.

2026-06-30

Building a Queue Dashboard: Real-Time Monitoring UX Patterns

Learn the UX patterns behind building a real-time queue monitoring dashboard for BullMQ and BeeQueue. From metric cards to time-series charts, polling strategies to empty states, discover the architecture that powers effective queue observability.