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 listeners, WebSocket push, job lifecycle visualization, and sliding-window metrics aggregation.

Introduction — The Polling Tax

Most queue dashboards work the same way: a setInterval fires every N seconds, calls getJobCounts(), getWorkers(), and getMetrics(), then renders the result. It works — but it has real costs:

• Stale data. You always see N-seconds-old state. A job can fail and be retried between polls without you ever noticing.
• Redis load. Every poll cycle scans Redis keys, runs Lua scripts, and pulls full queue state — multiplied across every connected user.
• Missed transitions. Short-lived jobs appear and disappear within a single polling interval. A job that's added, processed, and completed in 200ms is invisible to a 5-second poll cycle.
• Inefficient bandwidth. You fetch the full response payload every cycle even when nothing changed.

The alternative is event-driven monitoring. Instead of asking Redis "what happened?", you listen for the answer the instant it happens. BullMQ emits native events at every stage of a job's lifecycle: waiting, delayed, active, progress, completed, failed, stalled, drained. Every transition, in real time, delivered via Redis Streams.

This post walks through building a complete event-driven pipeline: capturing BullMQ events with QueueEvents, bridging them to WebSocket push, monitoring worker health, visualizing job lifecycle transitions, and aggregating real-time metrics — all with concrete TypeScript code.

---

Section 1: Understanding BullMQ's Event Architecture

BullMQ provides three layers of event sources, each with a distinct role:

1. Worker Events (Local)

A Worker instance emits events for jobs it processes itself. These are process-local — if you run three worker processes, each sees only its own jobs:

const worker = new Worker('email', async (job) => {
  await job.updateProgress(50);
  return { sent: true };
}, { connection });

worker.on('completed', (job, result) => {
  // Only fires for jobs THIS worker processed
  console.log(`[local] Job ${job.id} completed`, result);
});

worker.on('failed', (job, error) => {
  console.log(`[local] Job ${job.id} failed:`, error.message);
});

Worker events are useful for logging and local instrumentation, but they can't power a centralized dashboard.

2. QueueEvents (Global, Cross-Process)

QueueEvents uses Redis Streams under the hood. It consumes from a dedicated stream key (bull:{queueName}:events) and receives every event from every worker, process, or client connected to that queue. This is the backbone of event-driven dashboarding:

import { QueueEvents } from 'bullmq';
import { Redis } from 'ioredis';

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

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

queueEvents.on('completed', ({ jobId, returnvalue }) => {
  console.log(`[global] Job ${jobId} completed:`, returnvalue);
});

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

queueEvents.on('active', ({ jobId, prev }) => {
  console.log(`[global] Job ${jobId} moved from ${prev} to active`);
});

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

queueEvents.on('waiting', ({ jobId }) => {
  console.log(`[global] Job ${jobId} is waiting`);
});

queueEvents.on('delayed', ({ jobId, delay }) => {
  console.log(`[global] Job ${jobId} delayed by ${delay}ms`);
});

Critical rule: Always use a separate Redis connection for QueueEvents. It uses blocking XREADGROUP under the hood with different lifecycle and retry semantics from Worker and Queue connections. Sharing connections causes stalled readers and unexpected reconnection behavior.

3. Global Events (Stream Prefix)

Events in the stream are also available with a global: prefix (completed → global:completed). This is an alternative namespace for consumers that want to listen to "everything" without filtering by event type — but the QueueEvents.on() approach is more explicit and type-safe.

Event Namespace Hierarchy

bull:myqueue:events               ← Redis Stream key
├── completed                     ← event type
├── failed
├── progress
├── active
├── waiting
├── delayed
├── stalled
├── drained
├── removed
└── global:completed              ← global prefix variants
    global:failed
    global:progress
    ...

Redis Streams give you guaranteed delivery (consumer groups track what each consumer has read), auto-trim (configurable maxLen), and replay (re-read from any point in the stream). These are critical for production dashboard reliability.

---

Section 2: Building the Event Bridge — QueueEvents to WebSocket Push

QueueEvents emits events in the Node.js process, but your dashboard runs in the browser. The solution is a bridge service that listens to QueueEvents and fans events out to connected WebSocket clients.

Architecture

BullMQ Workers (process jobs)
  │
  ▼
QueueEvents (Redis Streams consumer)
  │
  ▼
Event Bridge Service (Node.js daemon)
  │
  ├──► WebSocket Server ──► Browser Dashboard
  │
  └──► Metrics Aggregator (in-memory sliding windows)

Full Bridge Implementation

import { WebSocketServer, WebSocket } from 'ws';
import { QueueEvents } from 'bullmq';
import { Redis } from 'ioredis';

interface BridgeConfig {
  queueName: string;
  wsPort: number;
  redisUrl: string;
}

interface WsMessage {
  type:
    | 'job:completed'
    | 'job:failed'
    | 'job:progress'
    | 'job:active'
    | 'job:waiting'
    | 'job:delayed'
    | 'job:stalled';
  payload: Record<string, unknown>;
  timestamp: number;
}

export class QueueEventBridge {
  private wss: WebSocketServer;
  private queueEvents: QueueEvents;
  private clients: Set<WebSocket> = new Set();

  constructor(config: BridgeConfig) {
    const eventsConnection = new Redis(config.redisUrl, {
      maxRetriesPerRequest: null,
      reconnectOnError: () => true,
    });

    this.wss = new WebSocketServer({ port: config.wsPort });
    this.queueEvents = new QueueEvents(config.queueName, {
      connection: eventsConnection,
      autorun: true,
    });

    this.setupWebSocket();
    this.setupQueueEvents();
  }

  private setupWebSocket(): void {
    this.wss.on('connection', (ws: WebSocket) => {
      this.clients.add(ws);
      ws.on('close', () => this.clients.delete(ws));
      ws.on('error', () => this.clients.delete(ws));
    });
  }

  private setupQueueEvents(): void {
    const emit = (event: string, payload: Record<string, unknown>) => {
      const message: WsMessage = {
        type: `job:${event}` as WsMessage['type'],
        payload,
        timestamp: Date.now(),
      };
      const serialized = JSON.stringify(message);
      for (const client of this.clients) {
        if (client.readyState === WebSocket.OPEN) {
          client.send(serialized);
        }
      }
    };

    this.queueEvents.on('waiting', ({ jobId }) => emit('waiting', { jobId }));
    this.queueEvents.on('active', ({ jobId, prev }) =>
      emit('active', { jobId, prev }),
    );
    this.queueEvents.on('completed', ({ jobId, returnvalue }) =>
      emit('completed', { jobId, returnvalue }),
    );
    this.queueEvents.on('failed', ({ jobId, failedReason }) =>
      emit('failed', { jobId, failedReason }),
    );
    this.queueEvents.on('progress', ({ jobId, data }) =>
      emit('progress', { jobId, data }),
    );
    this.queueEvents.on('delayed', ({ jobId, delay }) =>
      emit('delayed', { jobId, delay }),
    );
    this.queueEvents.on('stalled', ({ jobId }) =>
      emit('stalled', { jobId }),
    );
  }

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

// Usage
const bridge = new QueueEventBridge({
  queueName: 'email-queue',
  wsPort: 3002,
  redisUrl: process.env.REDIS_URL!,
});

Design decisions:

• One QueueEvents listener per queue, not one per worker — keeps the architecture centralized
• WebSocket server can run as a separate process for horizontal scaling (use Redis pub/sub to relay events between bridge instances)
• Auto-reconnect: QueueEvents reconnects on Redis disconnects, but you must handle re-subscribing the WebSocket connections

Client-Side Consumption

On the dashboard frontend, connect to the WebSocket and dispatch events to the UI:

const ws = new WebSocket('ws://localhost:3002');

ws.onmessage = (event) => {
  const message = JSON.parse(event.data);

  switch (message.type) {
    case 'job:completed':
      updateJobList(message.payload.jobId, 'completed');
      incrementMetric('completed');
      break;
    case 'job:failed':
      updateJobList(message.payload.jobId, 'failed');
      incrementMetric('failed');
      showNotification(message.payload.jobId);
      break;
    case 'job:active':
      updateJobList(message.payload.jobId, 'active');
      break;
    case 'job:progress':
      updateProgress(message.payload.jobId, message.payload.data);
      break;
  }
};

---

Section 3: Worker Monitoring — Tracking Status, Concurrency, and Throughput

BullMQ's built-in queue.getWorkers() gives you a snapshot of connected workers at a single point in time. For real-time monitoring, you need a continuous heartbeat model:

• Worker goes online → track it
• Worker processes jobs → update processed count and concurrency
• Worker goes silent → mark it as offline

BullMQ doesn't natively emit "worker connected" or "worker disconnected" events, but you can build this with two patterns working together:

1. Heartbeat keys with TTL — each worker writes its status to a Redis hash with a 30-second TTL; expiration equals disconnection
2. QueueEvents activity tracking — infer worker activity from active / completed / failed event deltas

Worker-Side Heartbeat Publisher

Run this inside every worker process to publish its status:

import { Worker, Job } from 'bullmq';
import { Redis } from 'ioredis';
import { randomUUID } from 'crypto';

export function createObservableWorker(
  queueName: string,
  processor: (job: Job) => Promise<unknown>,
  options: { connection: Redis; concurrency?: number },
): Worker {
  const workerId = `worker-${randomUUID().slice(0, 8)}`;
  const monitorRedis = new Redis(options.connection.options);
  const HEARTBEAT_KEY = `bullmq-monitor:workers:${queueName}:${workerId}`;
  const HEARTBEAT_INTERVAL = 10_000; // 10 seconds

  // Periodic heartbeat
  const heartbeatTimer = setInterval(async () => {
    try {
      await monitorRedis.hset(HEARTBEAT_KEY, {
        id: workerId,
        queueName,
        lastHeartbeat: Date.now().toString(),
        maxConcurrency: String(options.concurrency ?? 1),
        currentConcurrency: '0',
      });
      await monitorRedis.expire(HEARTBEAT_KEY, 30);
    } catch {
      // Heartbeat failure — process might be shutting down
    }
  }, HEARTBEAT_INTERVAL);

  const worker = new Worker(queueName, processor, {
    connection: options.connection,
    concurrency: options.concurrency ?? 1,
  });

  worker.on('active', async () => {
    await monitorRedis.hincrby(HEARTBEAT_KEY, 'currentConcurrency', 1);
  });

  worker.on('completed', async () => {
    await monitorRedis.hincrby(HEARTBEAT_KEY, 'currentConcurrency', -1);
    await monitorRedis.hincrby(HEARTBEAT_KEY, 'jobsProcessed', 1);
  });

  worker.on('failed', async () => {
    await monitorRedis.hincrby(HEARTBEAT_KEY, 'currentConcurrency', -1);
    await monitorRedis.hincrby(HEARTBEAT_KEY, 'jobsProcessed', 1);
  });

  // Cleanup on close
  const originalClose = worker.close.bind(worker);
  worker.close = async () => {
    clearInterval(heartbeatTimer);
    await monitorRedis.del(HEARTBEAT_KEY);
    await monitorRedis.quit();
    return originalClose();
  };

  return worker;
}

Dashboard-Side Worker Monitor

On the dashboard backend, a WorkerMonitor class reads heartbeat keys and tracks state:

import { Redis } from 'ioredis';

interface WorkerState {
  id: string;
  queueName: string;
  lastHeartbeat: number;
  jobsProcessed: number;
  currentConcurrency: number;
  maxConcurrency: number;
  status: 'active' | 'idle' | 'offline';
}

export class WorkerMonitor {
  private redis: Redis;
  private workers = new Map<string, WorkerState>();
  private readonly HEARTBEAT_TTL = 30;

  constructor(redisUrl: string) {
    this.redis = new Redis(redisUrl, { maxRetriesPerRequest: null });
  }

  async syncFromRedis(): Promise<void> {
    const pattern = 'bullmq-monitor:workers:*';
    const keys = await this.scanKeys(pattern);
    const currentWorkerIds = new Set<string>();

    for (const key of keys) {
      const raw = await this.redis.hgetall(key);
      if (!raw || !raw.id) continue;

      const workerId = raw.id;
      currentWorkerIds.add(workerId);

      this.workers.set(workerId, {
        id: workerId,
        queueName: raw.queueName || 'unknown',
        lastHeartbeat: parseInt(raw.lastHeartbeat || '0', 10),
        jobsProcessed: parseInt(raw.jobsProcessed || '0', 10),
        currentConcurrency: parseInt(raw.currentConcurrency || '0', 10),
        maxConcurrency: parseInt(raw.maxConcurrency || '1', 10),
        status: this.isRecent(parseInt(raw.lastHeartbeat || '0', 10))
          ? 'active'
          : 'idle',
      });
    }

    // Mark missing workers as offline
    for (const [id, state] of this.workers) {
      if (!currentWorkerIds.has(id)) {
        this.workers.set(id, { ...state, status: 'offline' });
      }
    }
  }

  getWorkers(): WorkerState[] {
    return Array.from(this.workers.values());
  }

  getActiveWorkerCount(): number {
    return [...this.workers.values()].filter(
      (w) => w.status === 'active',
    ).length;
  }

  private isRecent(timestamp: number): boolean {
    return Date.now() - timestamp < this.HEARTBEAT_TTL * 1000;
  }

  private async scanKeys(pattern: string): Promise<string[]> {
    const keys: string[] = [];
    let cursor = '0';
    do {
      const [next, batch] = await this.redis.scan(
        cursor,
        'MATCH',
        pattern,
        'COUNT',
        100,
      );
      cursor = next;
      keys.push(...batch);
    } while (cursor !== '0');
    return keys;
  }

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

This gives you a live worker overview: which workers are active, how many jobs each has processed, and their current concurrency utilization. Missing heartbeats (worker crash, network partition) are detected after 30 seconds and marked offline.

---

Section 4: Job Lifecycle Visualization — Mapping Every Transition

BullMQ jobs pass through a well-defined state machine:

[waiting] ──► [delayed] ──► [waiting] ──► [active] ──► [completed]
     │                                               │
     │                                               │
     └──► [waiting-children] ──► [waiting]           └──► [failed] ──► [waiting] (retry)
                                                                   │
                                                                   └──► [failed] (terminal)

Every transition emits a QueueEvents event. By tracking these events per job, you can reconstruct the full lifecycle of every job — including wait duration (time from waiting to active), processing duration (time from active to completed/failed), and total turnaround time.

JobLifecycleTracker Implementation

import { QueueEvents } from 'bullmq';

interface LifecycleEvent {
  state:
    | 'waiting'
    | 'delayed'
    | 'active'
    | 'completed'
    | 'failed'
    | 'progress'
    | 'stalled';
  timestamp: number;
  data?: unknown;
}

interface JobLifecycle {
  jobId: string;
  events: LifecycleEvent[];
  waitDuration?: number;
  processingDuration?: number;
  totalDuration?: number;
  attempts: number;
}

export class JobLifecycleTracker {
  private jobs = new Map<string, JobLifecycle>();
  private readonly MAX_JOBS = 1000;

  attach(queueEvents: QueueEvents): void {
    queueEvents.on('waiting', ({ jobId }) => {
      this.ensure(jobId);
      this.jobs
        .get(jobId)!
        .events.push({ state: 'waiting', timestamp: Date.now() });
    });

    queueEvents.on('delayed', ({ jobId, delay }) => {
      this.ensure(jobId);
      this.jobs.get(jobId)!.events.push({
        state: 'delayed',
        timestamp: Date.now(),
        data: { delay },
      });
    });

    queueEvents.on('active', ({ jobId }) => {
      this.ensure(jobId);
      const job = this.jobs.get(jobId)!;
      job.events.push({ state: 'active', timestamp: Date.now() });
      job.attempts++;

      // Calculate wait duration from the most recent waiting/delayed event
      const previousWait = [...job.events]
        .reverse()
        .find(
          (e) => e.state === 'waiting' || e.state === 'delayed',
        );
      if (previousWait) {
        job.waitDuration = Date.now() - previousWait.timestamp;
      }
    });

    queueEvents.on('progress', ({ jobId, data }) => {
      this.ensure(jobId);
      this.jobs.get(jobId)!.events.push({
        state: 'progress',
        timestamp: Date.now(),
        data,
      });
    });

    queueEvents.on('completed', ({ jobId, returnvalue }) => {
      this.ensure(jobId);
      const job = this.jobs.get(jobId)!;
      const now = Date.now();
      job.events.push({
        state: 'completed',
        timestamp: now,
        data: returnvalue,
      });

      const activeEvent = [...job.events]
        .reverse()
        .find((e) => e.state === 'active');
      if (activeEvent) {
        job.processingDuration = now - activeEvent.timestamp;
      }

      const firstEvent = job.events[0];
      if (firstEvent) {
        job.totalDuration = now - firstEvent.timestamp;
      }

      this.evictIfNeeded();
    });

    queueEvents.on('failed', ({ jobId, failedReason }) => {
      this.ensure(jobId);
      const job = this.jobs.get(jobId)!;
      const now = Date.now();
      job.events.push({
        state: 'failed',
        timestamp: now,
        data: failedReason,
      });

      const activeEvent = [...job.events]
        .reverse()
        .find((e) => e.state === 'active');
      if (activeEvent) {
        job.processingDuration = now - activeEvent.timestamp;
      }
    });

    queueEvents.on('stalled', ({ jobId }) => {
      this.ensure(jobId);
      this.jobs
        .get(jobId)!
        .events.push({ state: 'stalled', timestamp: Date.now() });
    });
  }

  getLifecycle(jobId: string): JobLifecycle | undefined {
    return this.jobs.get(jobId);
  }

  getAllLifecycles(): JobLifecycle[] {
    return Array.from(this.jobs.values());
  }

  getJobsInState(state: LifecycleEvent['state']): string[] {
    return Array.from(this.jobs.entries())
      .filter(
        ([, job]) => job.events[job.events.length - 1]?.state === state,
      )
      .map(([id]) => id);
  }

  private ensure(jobId: string): void {
    if (!this.jobs.has(jobId)) {
      this.jobs.set(jobId, { jobId, events: [], attempts: 0 });
    }
  }

  private evictIfNeeded(): void {
    if (this.jobs.size <= this.MAX_JOBS) return;

    const entries = Array.from(this.jobs.entries());
    entries.sort(([, a], [, b]) => {
      const aLast = a.events[a.events.length - 1]?.timestamp ?? 0;
      const bLast = b.events[b.events.length - 1]?.timestamp ?? 0;
      return aLast - bLast;
    });

    const toRemove = entries.slice(0, this.jobs.size - this.MAX_JOBS);
    for (const [id] of toRemove) {
      this.jobs.delete(id);
    }
  }
}

Dashboard UI Ideas for Lifecycle Visualization

• Job timeline view — Gantt-chart-like horizontal bars showing wait → active → complete per job, with color coding for each stage
• Live state counter — real-time count of jobs in waiting, active, delayed, completed, failed fed by event counters (not polling getJobCounts())
• Per-job state history panel — collapsible timeline showing every transition with timestamps and durations, including progress updates
• Lifecycle flow diagram — Sankey diagram showing how many jobs flow through each state transition, helping identify bottlenecks

---

Section 5: Metrics Aggregation from Events — Sliding Windows and Real-Time Counters

BullMQ provides built-in per-minute metrics via queue.getMetrics('completed'), but these only update at 1-minute granularity and don't include latency percentiles. For a live dashboard, you need sub-minute resolution and richer statistics.

The approach: aggregate your own metrics from the event stream using sliding window counters.

Sliding Window Pattern

Divide time into fixed-size buckets (e.g., 10 seconds). Each bucket counts events independently. When querying, total the last N buckets. Old buckets are pruned automatically.

import { QueueEvents } from 'bullmq';

interface WindowBucket {
  timestamp: number;
  completed: number;
  failed: number;
  totalProcessingTimeMs: number;
  processingTimeSamples: number[];
}

export class SlidingWindowMetrics {
  private buckets: WindowBucket[] = [];
  private readonly BUCKET_SIZE_MS = 10_000; // 10-second buckets
  private readonly WINDOW_SIZE_MS = 300_000; // 5-minute sliding window
  private activeJobCount = 0;
  private readonly MAX_LATENCY_SAMPLES = 1000;

  attach(queueEvents: QueueEvents): void {
    const activeTimestamps = new Map<string, number>();

    queueEvents.on('active', ({ jobId }) => {
      this.activeJobCount++;
      activeTimestamps.set(jobId, Date.now());
    });

    queueEvents.on('completed', ({ jobId }) => {
      this.activeJobCount--;
      this.recordEvent('completed', 1);
      const started = activeTimestamps.get(jobId);
      if (started) {
        this.recordLatency(Date.now() - started);
        activeTimestamps.delete(jobId);
      }
    });

    queueEvents.on('failed', ({ jobId }) => {
      this.activeJobCount--;
      this.recordEvent('failed', 1);
      const started = activeTimestamps.get(jobId);
      if (started) {
        this.recordLatency(Date.now() - started);
        activeTimestamps.delete(jobId);
      }
    });
  }

  private getOrCreateBucket(): WindowBucket {
    const now = Date.now();
    const bucketStart =
      Math.floor(now / this.BUCKET_SIZE_MS) * this.BUCKET_SIZE_MS;

    let bucket = this.buckets[this.buckets.length - 1];
    if (!bucket || bucket.timestamp !== bucketStart) {
      bucket = {
        timestamp: bucketStart,
        completed: 0,
        failed: 0,
        totalProcessingTimeMs: 0,
        processingTimeSamples: [],
      };
      this.buckets.push(bucket);
      this.prune();
    }
    return bucket;
  }

  private recordEvent(type: 'completed' | 'failed', count: number): void {
    const bucket = this.getOrCreateBucket();
    if (type === 'completed') bucket.completed += count;
    else bucket.failed += count;
  }

  private recordLatency(ms: number): void {
    const bucket = this.getOrCreateBucket();
    bucket.totalProcessingTimeMs += ms;
    if (bucket.processingTimeSamples.length < this.MAX_LATENCY_SAMPLES) {
      bucket.processingTimeSamples.push(ms);
    }
  }

  private prune(): void {
    const cutoff = Date.now() - this.WINDOW_SIZE_MS;
    while (
      this.buckets.length > 0 &&
      this.buckets[0].timestamp < cutoff
    ) {
      this.buckets.shift();
    }
  }

  getMetrics(): {
    throughputPerSec: number;
    failureRate: number;
    avgProcessingTimeMs: number;
    p50ProcessingTimeMs: number;
    p95ProcessingTimeMs: number;
    p99ProcessingTimeMs: number;
    activeJobs: number;
  } {
    const totalCompleted = this.buckets.reduce(
      (s, b) => s + b.completed,
      0,
    );
    const totalFailed = this.buckets.reduce((s, b) => s + b.failed, 0);
    const totalJobs = totalCompleted + totalFailed;
    const totalTimeSec = this.WINDOW_SIZE_MS / 1000;

    const allLatencies = this.buckets
      .flatMap((b) => b.processingTimeSamples)
      .sort((a, b) => a - b);
    const avgMs =
      allLatencies.length > 0
        ? this.buckets.reduce((s, b) => s + b.totalProcessingTimeMs, 0) /
          allLatencies.length
        : 0;

    const percentile = (p: number): number => {
      if (allLatencies.length === 0) return 0;
      const idx = Math.ceil((p / 100) * allLatencies.length) - 1;
      return allLatencies[Math.max(0, idx)] ?? 0;
    };

    return {
      throughputPerSec: totalTimeSec > 0 ? totalJobs / totalTimeSec : 0,
      failureRate: totalJobs > 0 ? totalFailed / totalJobs : 0,
      avgProcessingTimeMs: avgMs,
      p50ProcessingTimeMs: percentile(50),
      p95ProcessingTimeMs: percentile(95),
      p99ProcessingTimeMs: percentile(99),
      activeJobs: this.activeJobCount,
    };
  }
}

This gives you live throughput (jobs/sec), failure rate (as a decimal), latency percentiles (P50, P95, P99), and active job count — all updated in real time as events arrive.

Design Considerations

• In-memory aggregation resets on process restart. For production, pair with Redis-backed counters for durability and a time-series DB (ClickHouse, TimescaleDB) for historical queries.
• Active job counter drift. If a worker crashes mid-job, the counter stays inflated. Mitigate with periodic reconciliation using queue.getActiveCount() as a baseline correction.
• Bucket sizing. 10-second buckets with a 5-minute window gives 30 data points — enough for smooth visualizations without excessive memory.

---

Section 6: How QueueHub Implements These Patterns

QueueHub's architecture mirrors the event bridge pattern from Section 2, extended for multi-backend support (BullMQ and BeeQueue) and multi-org isolation.

The Three-Tier Event Pipeline

1. Connection layer — per-user Redis connections (standard, TLS, Valkey, AWS ElastiCache) managed with lazy connect and reconnect
2. Event bridge service — a centralized daemon that creates QueueEvents listeners for each active org's queues, bridging events to...
3. WebSocket distribution — per-org WebSocket channels; events are org-scoped so tenants never see each other's jobs

Agent Tunneling for Private Redis

When Redis runs behind a firewall, QueueHub's agent runs the event bridge on the customer's network and relays events over the tunnel WebSocket back to the cloud service. The heartbeat pattern from Section 3 runs inside the agent process.

Worker Monitoring

QueueHub uses the heartbeat-with-TTL pattern, storing worker metadata under qh:workers:{orgId}:{queueName}:{workerId} with a 30-second TTL. The live worker view updates via the same WebSocket channel as job events.

Job Lifecycle Visualization

The JobLifecycleTracker pattern feeds the job detail panel. Each job shows a timeline highlighting time in waiting vs active, total duration, and retry attempts. Jobs that have been retried show the full history of attempts with their failure reasons.

Metrics

Sliding-window in-memory aggregation handles the "last 5 minutes" real-time view. Historical data is persisted to Postgres for "last 24 hours" time-series charts. QueueHub chose this over BullMQ's built-in metrics because built-in metrics:

• Aggregate at 1-minute granularity (too coarse for real-time)
• Track only completed/failed counts
• Don't provide latency percentiles

Architecture Overview

┌──────────────────────────────────────────────────────┐
│              QueueHub Cloud (Node.js/Bun)             │
│                                                       │
│  ┌──────────┐   ┌────────────────┐   ┌─────────────┐ │
│  │ Hono API │   │ Event Bridge   │   │ WebSocket   │ │
│  │ Server   │   │ Daemon         │   │ Distributor │ │
│  │          │   │                │   │             │ │
│  │ REST     │   │ QueueEvents ×N │   │ Per-org WSS │ │
│  │ endpoints│   │ SlidingWindow  │   │ channels    │ │
│  └────┬─────┘   │ WorkerMonitor  │   └──────┬──────┘ │
│       │         └───────┬────────┘          │        │
│       │                 │                   │        │
│       ▼                 ▼                   ▼        │
│  ┌─────────────────────────────────────────────┐     │
│  │          Redis (per org/connection)         │     │
│  │  ┌─────┐ ┌──────┐ ┌────────┐ ┌──────────┐  │     │
│  │  │Jobs │ │Queues│ │Workers │ │Metrics   │  │     │
│  │  │data │ │meta  │ │heart   │ │buckets   │  │     │
│  │  │     │ │      │ │beat    │ │          │  │     │
│  │  └─────┘ └──────┘ └────────┘ └──────────┘  │     │
│  └─────────────────────────────────────────────┘     │
│               │                                      │
│               │  (agent tunnel for private Redis)    │
│               ▼                                      │
│  ┌────────────────────────────────┐                  │
│  │  Agent (on customer network)   │                  │
│  │  ┌──────────────────────────┐  │                  │
│  │  │ QueueEvents + Heartbeat  │  │                  │
│  │  │ Relay → WSS              │  │                  │
│  │  └──────────────────────────┘  │                  │
│  └────────────────────────────────┘                  │
└──────────────────────────────────────────────────────┘

---

Section 7: Production Considerations

QueueEvents Stream Backpressure

The default maxLen for event streams is ~10,000 events. In high-throughput queues (100+ jobs/sec), events can be trimmed before the dashboard reads them. Solutions:

// Increase maxLen during QueueEvents instantiation
const queueEvents = new QueueEvents('my-queue', {
  connection,
  streams: {
    events: {
      maxLen: 100_000, // Larger buffer for high throughput
    },
  },
});

For very high throughput, run a dedicated consumer that offloads events to a time-series database and keep the stream window small.

Separate Redis Connections

Never share a Redis connection between a Worker/Queue and QueueEvents. They have different blocking semantics — QueueEvents uses XREADGROUP with blocking, while Workers use BRPOPLPUSH / BLPOP. Sharing causes one to starve the other.

Connection Pool Management

QueueEvents uses one blocking connection per queue. For 50 queues, you need 50 dedicated connections plus the main connection. Use a connection pool manager like ioredis Cluster or a dedicated connection factory.

Scaling the WebSocket Layer

When running multiple bridge instances behind a load balancer, a client's WebSocket might connect to Bridge A while QueueEvents events arrive at Bridge B. Use Redis pub/sub as a relay:

QueueEvents → Bridge A → [Redis Pub/Sub channel] → Bridge B, Bridge C → WS clients

Every bridge instance subscribes to the same pub/sub channel. When any bridge receives a QueueEvents event, it publishes to the channel, and all bridges — including the one that published — push to their local WebSocket clients.

Memory Management

JobLifecycleTracker grows unbounded. Set limits:

• Maximum tracked jobs (evict oldest completed/failed)
• TTL-based eviction for completed jobs (e.g., drop after 5 minutes)
• Offload completed job lifecycles to disk/DB after a threshold

Reconnection Semantics

When Redis fails over, QueueEvents reconnects automatically (via XREADGROUP with > semantics). The stream consumer group tracks the last read ID, so no events are double-processed or missed — as long as the stream hasn't been trimmed past the last read point.

Security in Multi-Tenant Dashboards

Org-scope every event. Never broadcast a job event from Org A to Org B's WebSocket. Use per-org QueueEvents instances keyed by each org's Redis connection settings:

class OrgEventBridge {
  private bridges = new Map<string, QueueEventBridge>();

  async connectOrg(orgId: string, redisConfig: RedisConfig) {
    const bridge = new QueueEventBridge({
      queueName: orgConfig.queueName,
      wsPort: 0, // shared WebSocket server with per-channel routing
      redisUrl: redisConfig.url,
    });
    this.bridges.set(orgId, bridge);
  }

  // WebSocket router checks orgId before dispatching
}

Error Handling

Always attach an error listener to QueueEvents. Unhandled errors crash the event bridge:

queueEvents.on('error', (err) => {
  console.error('[QueueEvents] Error:', err);
  // Trigger alert, attempt reconnect
});

Comparison: Polling vs Event-Driven

Aspect	Polling-Based Dashboard	Event-Driven Dashboard
Data freshness	N-second stale (poll interval)	Sub-second (event arrival)
Redis load	High (SCAN, getJobCounts, getWorkers per poll)	Low (stream reads only)
Missed jobs	Yes (short jobs between polls)	No (every transition captured)
Implementation complexity	Low	Medium-High
Horizontal scaling	Stateless (more REST servers)	Stateful (fan events across instances)
Historical replay	Not possible	Possible from stream (if not trimmed)

---

Conclusion — From Snapshot to Stream

Polling gave us snapshots; events give us a live stream of everything happening in the queue. The shift from "what's the current state?" to "what just happened?" unlocks real-time observability:

• Spot failure spikes the instant they occur, not N seconds later
• Watch jobs flow through each lifecycle stage with sub-second latency
• Track worker health and concurrency as it changes
• Query latency percentiles and throughput on demand

BullMQ's event system provides all the primitives — Worker events for local instrumentation, QueueEvents for cross-process visibility, and Redis Streams for reliable delivery. The rest is plumbing, but it's well-understood plumbing that any Node.js team can implement.

Try It Yourself

Start with a single QueueEvents listener, bridge it to a WebSocket, and watch your queue come alive. The code examples in this post give you a complete foundation.

For a ready-to-use solution, QueueHub implements this event-driven architecture for every connected queue backend — BullMQ, BeeQueue, or any Redis-backed queue. Connect your Redis, and your dashboard updates in real time with live worker monitoring, job lifecycle timelines, and sliding-window metrics.

Appendix: BullMQ Event Reference

Event	Source	Payload	When It Fires
waiting	QueueEvents	{ jobId }	Job added to waiting list
active	Worker, QueueEvents	{ jobId, prev }	Job started processing
completed	Worker, QueueEvents	{ jobId, returnvalue }	Job succeeded
failed	Worker, QueueEvents	{ jobId, failedReason }	Job failed
progress	Worker, QueueEvents	{ jobId, data }	job.updateProgress() called
delayed	QueueEvents	{ jobId, delay }	Job moved to delayed set
stalled	QueueEvents	{ jobId }	Worker lost lock on job
drained	Worker, QueueEvents	{}	No more jobs in queue
removed	QueueEvents	{ jobId }	Job manually removed
error	Worker, QueueEvents	Error	Connection error or internal failure