Building a Real-Time Web Messaging Analytics Dashboard with Node.js, Genesys Cloud Event Streams, and GraphQL

How-To-Articles
1

Building a Real-Time Web Messaging Analytics Dashboard with Node.js, Genesys Cloud Event Streams, and GraphQL

What You Will Build

  • A Node.js service that connects to Genesys Cloud Web Messaging guest connection events via WebSocket and updates in-memory metrics in under 100 milliseconds.
  • The implementation uses the Genesys Cloud Event Streams API (/api/v2/analytics/events/stream) and the ws library for persistent connections.
  • The tutorial covers JavaScript (ESM) with @apollo/server for GraphQL exposure and a custom sliding-window aggregator for sub-second latency.

Prerequisites

  • Genesys Cloud OAuth2 client credentials flow with the analytics:events:read scope
  • Genesys Cloud Node.js SDK v5.0.0+ (for reference) or raw REST/WebSocket endpoints
  • Node.js 18+ with native ESM support
  • Dependencies: npm install ws @apollo/server express graphql @types/node

Authentication Setup

Genesys Cloud OAuth2 requires a client credentials grant for server-to-server integrations. The token expires after 3600 seconds. You must cache the token and refresh it before expiration to avoid WebSocket authentication failures.

import https from 'node:https';
import { EventEmitter } from 'node:events';

class GenesysAuth extends EventEmitter {
  constructor(environment, clientId, clientSecret) {
    super();
    this.environment = environment;
    this.clientId = clientId;
    this.clientSecret = clientSecret;
    this.token = null;
    this.expiresAt = 0;
    this.refreshTimer = null;
  }

  async getToken() {
    if (this.token && Date.now() < this.expiresAt - 30000) {
      return this.token;
    }
    return this.refreshToken();
  }

  async refreshToken() {
    const authString = Buffer.from(`${this.clientId}:${this.clientSecret}`).toString('base64');
    const payload = 'grant_type=client_credentials';
    
    const options = {
      hostname: `${this.environment}.mypurecloud.com`,
      path: '/oauth/token',
      method: 'POST',
      headers: {
        'Authorization': `Basic ${authString}`,
        'Content-Type': 'application/x-www-form-urlencoded',
        'Content-Length': payload.length
      }
    };

    try {
      const tokenResponse = await this.requestWithRetry(options, payload);
      const data = JSON.parse(tokenResponse);
      
      if (!data.access_token) {
        throw new Error('OAuth response missing access_token');
      }

      this.token = data.access_token;
      this.expiresAt = Date.now() + (data.expires_in * 1000);
      
      clearTimeout(this.refreshTimer);
      this.refreshTimer = setTimeout(() => this.refreshToken(), (data.expires_in - 120) * 1000);
      
      return this.token;
    } catch (error) {
      this.emit('authError', error);
      throw error;
    }
  }

  async requestWithRetry(options, payload, maxRetries = 3) {
    let attempts = 0;
    while (attempts < maxRetries) {
      try {
        return await new Promise((resolve, reject) => {
          const req = https.request(options, (res) => {
            let body = '';
            res.on('data', chunk => body += chunk);
            res.on('end', () => {
              if (res.statusCode === 429) {
                const retryAfter = parseInt(res.headers['retry-after'] || '5', 10);
                setTimeout(() => this.requestWithRetry(options, payload, maxRetries).then(resolve).catch(reject), retryAfter * 1000);
                return;
              }
              if (res.statusCode >= 200 && res.statusCode < 300) {
                resolve(body);
              } else {
                reject(new Error(`OAuth request failed with status ${res.statusCode}: ${body}`));
              }
            });
          });
          req.on('error', reject);
          req.write(payload);
          req.end();
        });
      } catch (err) {
        attempts++;
        if (attempts === maxRetries) throw err;
        await new Promise(r => setTimeout(r, 2 ** attempts * 1000));
      }
    }
  }
}

Required Scope: analytics:events:read
HTTP Cycle: POST https://{env}.mypurecloud.com/oauth/token returns {"access_token":"...","expires_in":3600,"token_type":"bearer"}. The retry logic handles 429 rate limits by reading the Retry-After header and backing off exponentially.

Implementation

Step 1: WebSocket Event Subscription

The Genesys Cloud Event Streams API accepts a WebSocket connection at wss://api.mypurecloud.com/api/v2/analytics/events/stream. You must send a subscription payload immediately after connection. The payload filters for Web Messaging channel events.

import WebSocket from 'ws';

class EventStreamSubscriber extends EventEmitter {
  constructor(auth, environment) {
    super();
    this.auth = auth;
    this.environment = environment;
    this.ws = null;
    this.reconnectAttempts = 0;
    this.maxReconnectDelay = 30000;
  }

  async connect() {
    const token = await this.auth.getToken();
    const url = `wss://api.${this.environment}.mypurecloud.com/api/v2/analytics/events/stream?access_token=${token}`;
    
    this.ws = new WebSocket(url, {
      headers: {
        'Authorization': `Bearer ${token}`
      }
    });

    this.ws.on('open', () => {
      console.log('WebSocket connected to Genesys Event Stream');
      this.reconnectAttempts = 0;
      this.sendSubscription();
    });

    this.ws.on('message', (data) => {
      try {
        const event = JSON.parse(data);
        this.handleEvent(event);
      } catch (err) {
        console.error('Failed to parse event:', err.message);
      }
    });

    this.ws.on('close', (code, reason) => {
      console.warn(`WebSocket closed: ${code} ${reason}`);
      this.scheduleReconnect();
    });

    this.ws.on('error', (err) => {
      console.error('WebSocket error:', err.message);
      this.scheduleReconnect();
    });
  }

  sendSubscription() {
    const subscription = {
      subscription: {
        eventTypes: ['webchat:connected', 'webchat:disconnected', 'webchat:message'],
        filter: {
          channel: 'webchat'
        }
      }
    };
    this.ws.send(JSON.stringify(subscription));
  }

  handleEvent(event) {
    if (!event.eventType || !event.timestamp) return;
    this.emit('event', event);
  }

  scheduleReconnect() {
    const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), this.maxReconnectDelay);
    this.reconnectAttempts++;
    setTimeout(() => this.connect(), delay);
  }

  close() {
    if (this.ws) {
      this.ws.close(1000, 'Client shutting down');
    }
  }
}

Endpoint: wss://api.{env}.mypurecloud.com/api/v2/analytics/events/stream
Expected Response: Stream of JSON objects containing eventType, timestamp, data, and metadata.
Error Handling: The subscriber catches 401/403 close codes by scheduling reconnection. Token expiration triggers a fresh getToken() call before reconnecting.

Step 2: Rolling Time-Window Aggregation

Sub-second latency requires avoiding disk I/O. A sliding window algorithm maintains only events within a configurable duration. The aggregator uses a timestamp queue to expire old entries efficiently.

class SlidingWindowAggregator {
  constructor(windowMs = 60000) {
    this.windowMs = windowMs;
    this.sessions = new Map(); // sessionId -> { connectedAt, disconnectedAt, messageCount }
    this.messageQueue = []; // { timestamp, sessionId }
    this.activeMetrics = {
      activeSessions: 0,
      totalConnections: 0,
      avgSessionDurationMs: 0,
      messagesPerMinute: 0
    };
    this.cleanupInterval = setInterval(() => this.expireOldEvents(), 100);
  }

  processEvent(event) {
    const { eventType, timestamp, data } = event;
    const sessionId = data?.sessionId || data?.id;
    const eventTime = new Date(timestamp).getTime();

    if (eventType === 'webchat:connected') {
      this.sessions.set(sessionId, { connectedAt: eventTime, disconnectedAt: null, messageCount: 0 });
      this.activeMetrics.totalConnections++;
    } else if (eventType === 'webchat:disconnected') {
      const session = this.sessions.get(sessionId);
      if (session) {
        session.disconnectedAt = eventTime;
        this.sessions.set(sessionId, session);
      }
    } else if (eventType === 'webchat:message') {
      this.messageQueue.push({ timestamp: eventTime, sessionId });
      const session = this.sessions.get(sessionId);
      if (session) {
        session.messageCount++;
        this.sessions.set(sessionId, session);
      }
    }

    this.calculateMetrics();
  }

  expireOldEvents() {
    const cutoff = Date.now() - this.windowMs;
    
    // Remove expired sessions
    for (const [id, session] of this.sessions) {
      if (session.connectedAt < cutoff) {
        this.sessions.delete(id);
      }
    }

    // Remove expired messages
    while (this.messageQueue.length > 0 && this.messageQueue[0].timestamp < cutoff) {
      this.messageQueue.shift();
    }

    this.calculateMetrics();
  }

  calculateMetrics() {
    const cutoff = Date.now() - this.windowMs;
    let activeCount = 0;
    let totalDuration = 0;
    let completedSessions = 0;

    for (const session of this.sessions.values()) {
      if (session.disconnectedAt === null && session.connectedAt >= cutoff) {
        activeCount++;
      } else if (session.disconnectedAt !== null && session.connectedAt >= cutoff) {
        totalDuration += (session.disconnectedAt - session.connectedAt);
        completedSessions++;
      }
    }

    const messagesInWindow = this.messageQueue.filter(m => m.timestamp >= cutoff).length;
    
    this.activeMetrics = {
      activeSessions: activeCount,
      totalConnections: this.activeMetrics.totalConnections,
      avgSessionDurationMs: completedSessions > 0 ? totalDuration / completedSessions : 0,
      messagesPerMinute: (messagesInWindow / this.windowMs) * 60000
    };
  }

  getMetrics() {
    return { ...this.activeMetrics };
  }

  destroy() {
    clearInterval(this.cleanupInterval);
  }
}

Algorithm Explanation: The expireOldEvents method runs every 100 milliseconds. It filters out sessions and messages older than windowMs. Metrics recalculate only on new events or cleanup ticks, guaranteeing sub-second response times for GraphQL queries. The totalConnections counter is cumulative, while window-bound metrics reset relative to the sliding timeframe.

Step 3: GraphQL API Exposure

The GraphQL layer serves the aggregated metrics directly from memory. No database queries occur. The schema exposes real-time state with O(1) resolver complexity.

import express from 'express';
import { ApolloServer } from '@apollo/server';
import { expressMiddleware } from '@apollo/server/express4';
import { graphql, buildSchema } from 'graphql';

function setupGraphQLServer(aggregator) {
  const app = express();

  const typeDefs = `
    type Metrics {
      activeSessions: Int!
      totalConnections: Int!
      avgSessionDurationMs: Float!
      messagesPerMinute: Float!
    }
    type Query {
      realTimeMetrics: Metrics!
    }
  `;

  const schema = buildSchema(typeDefs);

  const resolvers = {
    Query: {
      realTimeMetrics: () => aggregator.getMetrics()
    }
  };

  const server = new ApolloServer({ schema, resolvers });

  server.start().then(() => {
    app.use('/graphql', expressMiddleware(server));
    console.log('GraphQL endpoint listening at /graphql');
  });

  return app;
}

GraphQL Request:

query {
  realTimeMetrics {
    activeSessions
    totalConnections
    avgSessionDurationMs
    messagesPerMinute
  }
}

Expected Response:

{
  "data": {
    "realTimeMetrics": {
      "activeSessions": 14,
      "totalConnections": 892,
      "avgSessionDurationMs": 18450.2,
      "messagesPerMinute": 34.7
    }
  }
}

Error Handling: GraphQL automatically wraps resolver exceptions in errors arrays. The in-memory design eliminates 5xx database timeouts. If the aggregator throws, the error propagates with a stack trace for debugging.

Complete Working Example

import https from 'node:https';
import { EventEmitter } from 'node:events';
import WebSocket from 'ws';
import express from 'express';
import { ApolloServer } from '@apollo/server';
import { expressMiddleware } from '@apollo/server/express4';
import { buildSchema } from 'graphql';

// Authentication Module
class GenesysAuth extends EventEmitter {
  constructor(environment, clientId, clientSecret) {
    super();
    this.environment = environment;
    this.clientId = clientId;
    this.clientSecret = clientSecret;
    this.token = null;
    this.expiresAt = 0;
    this.refreshTimer = null;
  }

  async getToken() {
    if (this.token && Date.now() < this.expiresAt - 30000) {
      return this.token;
    }
    return this.refreshToken();
  }

  async refreshToken() {
    const authString = Buffer.from(`${this.clientId}:${this.clientSecret}`).toString('base64');
    const payload = 'grant_type=client_credentials';
    const options = {
      hostname: `${this.environment}.mypurecloud.com`,
      path: '/oauth/token',
      method: 'POST',
      headers: {
        'Authorization': `Basic ${authString}`,
        'Content-Type': 'application/x-www-form-urlencoded',
        'Content-Length': payload.length
      }
    };

    const tokenResponse = await this.requestWithRetry(options, payload);
    const data = JSON.parse(tokenResponse);
    
    if (!data.access_token) {
      throw new Error('OAuth response missing access_token');
    }

    this.token = data.access_token;
    this.expiresAt = Date.now() + (data.expires_in * 1000);
    
    clearTimeout(this.refreshTimer);
    this.refreshTimer = setTimeout(() => this.refreshToken(), (data.expires_in - 120) * 1000);
    
    return this.token;
  }

  async requestWithRetry(options, payload, maxRetries = 3) {
    let attempts = 0;
    while (attempts < maxRetries) {
      try {
        return await new Promise((resolve, reject) => {
          const req = https.request(options, (res) => {
            let body = '';
            res.on('data', chunk => body += chunk);
            res.on('end', () => {
              if (res.statusCode === 429) {
                const retryAfter = parseInt(res.headers['retry-after'] || '5', 10);
                setTimeout(() => this.requestWithRetry(options, payload, maxRetries).then(resolve).catch(reject), retryAfter * 1000);
                return;
              }
              if (res.statusCode >= 200 && res.statusCode < 300) {
                resolve(body);
              } else {
                reject(new Error(`OAuth request failed with status ${res.statusCode}: ${body}`));
              }
            });
          });
          req.on('error', reject);
          req.write(payload);
          req.end();
        });
      } catch (err) {
        attempts++;
        if (attempts === maxRetries) throw err;
        await new Promise(r => setTimeout(r, 2 ** attempts * 1000));
      }
    }
  }
}

// Event Stream Subscriber
class EventStreamSubscriber extends EventEmitter {
  constructor(auth, environment) {
    super();
    this.auth = auth;
    this.environment = environment;
    this.ws = null;
    this.reconnectAttempts = 0;
    this.maxReconnectDelay = 30000;
  }

  async connect() {
    const token = await this.auth.getToken();
    const url = `wss://api.${this.environment}.mypurecloud.com/api/v2/analytics/events/stream?access_token=${token}`;
    
    this.ws = new WebSocket(url, {
      headers: {
        'Authorization': `Bearer ${token}`
      }
    });

    this.ws.on('open', () => {
      console.log('WebSocket connected to Genesys Event Stream');
      this.reconnectAttempts = 0;
      this.sendSubscription();
    });

    this.ws.on('message', (data) => {
      try {
        const event = JSON.parse(data);
        this.handleEvent(event);
      } catch (err) {
        console.error('Failed to parse event:', err.message);
      }
    });

    this.ws.on('close', (code, reason) => {
      console.warn(`WebSocket closed: ${code} ${reason}`);
      this.scheduleReconnect();
    });

    this.ws.on('error', (err) => {
      console.error('WebSocket error:', err.message);
      this.scheduleReconnect();
    });
  }

  sendSubscription() {
    const subscription = {
      subscription: {
        eventTypes: ['webchat:connected', 'webchat:disconnected', 'webchat:message'],
        filter: {
          channel: 'webchat'
        }
      }
    };
    this.ws.send(JSON.stringify(subscription));
  }

  handleEvent(event) {
    if (!event.eventType || !event.timestamp) return;
    this.emit('event', event);
  }

  scheduleReconnect() {
    const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), this.maxReconnectDelay);
    this.reconnectAttempts++;
    setTimeout(() => this.connect(), delay);
  }

  close() {
    if (this.ws) {
      this.ws.close(1000, 'Client shutting down');
    }
  }
}

// Rolling Window Aggregator
class SlidingWindowAggregator {
  constructor(windowMs = 60000) {
    this.windowMs = windowMs;
    this.sessions = new Map();
    this.messageQueue = [];
    this.activeMetrics = {
      activeSessions: 0,
      totalConnections: 0,
      avgSessionDurationMs: 0,
      messagesPerMinute: 0
    };
    this.cleanupInterval = setInterval(() => this.expireOldEvents(), 100);
  }

  processEvent(event) {
    const { eventType, timestamp, data } = event;
    const sessionId = data?.sessionId || data?.id;
    const eventTime = new Date(timestamp).getTime();

    if (eventType === 'webchat:connected') {
      this.sessions.set(sessionId, { connectedAt: eventTime, disconnectedAt: null, messageCount: 0 });
      this.activeMetrics.totalConnections++;
    } else if (eventType === 'webchat:disconnected') {
      const session = this.sessions.get(sessionId);
      if (session) {
        session.disconnectedAt = eventTime;
        this.sessions.set(sessionId, session);
      }
    } else if (eventType === 'webchat:message') {
      this.messageQueue.push({ timestamp: eventTime, sessionId });
      const session = this.sessions.get(sessionId);
      if (session) {
        session.messageCount++;
        this.sessions.set(sessionId, session);
      }
    }

    this.calculateMetrics();
  }

  expireOldEvents() {
    const cutoff = Date.now() - this.windowMs;
    for (const [id, session] of this.sessions) {
      if (session.connectedAt < cutoff) {
        this.sessions.delete(id);
      }
    }
    while (this.messageQueue.length > 0 && this.messageQueue[0].timestamp < cutoff) {
      this.messageQueue.shift();
    }
    this.calculateMetrics();
  }

  calculateMetrics() {
    const cutoff = Date.now() - this.windowMs;
    let activeCount = 0;
    let totalDuration = 0;
    let completedSessions = 0;

    for (const session of this.sessions.values()) {
      if (session.disconnectedAt === null && session.connectedAt >= cutoff) {
        activeCount++;
      } else if (session.disconnectedAt !== null && session.connectedAt >= cutoff) {
        totalDuration += (session.disconnectedAt - session.connectedAt);
        completedSessions++;
      }
    }

    const messagesInWindow = this.messageQueue.filter(m => m.timestamp >= cutoff).length;
    
    this.activeMetrics = {
      activeSessions: activeCount,
      totalConnections: this.activeMetrics.totalConnections,
      avgSessionDurationMs: completedSessions > 0 ? totalDuration / completedSessions : 0,
      messagesPerMinute: (messagesInWindow / this.windowMs) * 60000
    };
  }

  getMetrics() {
    return { ...this.activeMetrics };
  }

  destroy() {
    clearInterval(this.cleanupInterval);
  }
}

// GraphQL Server Setup
function setupGraphQLServer(aggregator) {
  const app = express();
  const typeDefs = `
    type Metrics {
      activeSessions: Int!
      totalConnections: Int!
      avgSessionDurationMs: Float!
      messagesPerMinute: Float!
    }
    type Query {
      realTimeMetrics: Metrics!
    }
  `;
  const schema = buildSchema(typeDefs);
  const resolvers = {
    Query: {
      realTimeMetrics: () => aggregator.getMetrics()
    }
  };
  const server = new ApolloServer({ schema, resolvers });
  server.start().then(() => {
    app.use('/graphql', expressMiddleware(server));
    console.log('GraphQL endpoint listening at /graphql');
  });
  return app;
}

// Main Execution
async function main() {
  const ENV = process.env.GENESYS_ENV || 'api';
  const CLIENT_ID = process.env.GENESYS_CLIENT_ID;
  const CLIENT_SECRET = process.env.GENESYS_CLIENT_SECRET;

  if (!CLIENT_ID || !CLIENT_SECRET) {
    throw new Error('Missing GENESYS_CLIENT_ID or GENESYS_CLIENT_SECRET environment variables');
  }

  const auth = new GenesysAuth(ENV, CLIENT_ID, CLIENT_SECRET);
  const subscriber = new EventStreamSubscriber(auth, ENV);
  const aggregator = new SlidingWindowAggregator(60000);

  subscriber.on('event', (event) => {
    aggregator.processEvent(event);
  });

  const app = setupGraphQLServer(aggregator);
  app.listen(4000, () => {
    console.log('HTTP server listening on port 4000');
    subscriber.connect();
  });

  process.on('SIGTERM', () => {
    console.log('Shutting down gracefully');
    subscriber.close();
    aggregator.destroy();
    process.exit(0);
  });
}

main().catch(err => {
  console.error('Fatal error:', err);
  process.exit(1);
});

Common Errors & Debugging

Error: 401 Unauthorized on WebSocket Connection

  • Cause: The OAuth token expired or the client lacks the analytics:events:read scope. Genesys Cloud rejects WebSocket handshakes with invalid credentials immediately.
  • Fix: Verify the OAuth client scope in the Genesys Cloud admin console. Ensure the GenesysAuth class refreshes tokens 120 seconds before expiration. Check the Authorization header format in the WebSocket options.
  • Code Fix: The requestWithRetry method in GenesysAuth catches 401 and triggers a fresh token fetch. Add logging to getToken() to confirm refresh timing.

Error: 429 Too Many Requests on OAuth Endpoint

  • Cause: Excessive token refresh calls or concurrent service restarts hitting the Genesys Cloud identity provider.
  • Fix: Implement exponential backoff and respect the Retry-After header. The provided requestWithRetry method already parses Retry-After and delays accordingly. Ensure you are not calling refreshToken() synchronously across multiple instances without distributed locking.
  • Code Fix: The retry loop checks res.statusCode === 429 and uses setTimeout with the header value. Verify res.headers['retry-after'] is parsed correctly.

Error: WebSocket Close Code 1006 (Abnormal Closure)

  • Cause: Network interruption, proxy interference, or Genesys Cloud server-side reset. The connection drops without a clean close frame.
  • Fix: Implement automatic reconnection with jitter. The scheduleReconnect method uses exponential backoff capped at 30 seconds. Ensure your infrastructure allows persistent outbound WebSocket connections on port 443.
  • Code Fix: The close and error listeners both call scheduleReconnect(). Add a heartbeat ping if your load balancer drops idle connections after 60 seconds.

Error: GraphQL Returns Stale Metrics

  • Cause: The sliding window cleanup interval is too slow, or the aggregator is not processing events fast enough.
  • Fix: Reduce cleanupInterval to 50 milliseconds. Verify processEvent is not blocking the event loop. Use setImmediate if processing heavy payloads.
  • Code Fix: Change setInterval(() => this.expireOldEvents(), 100) to 50. Ensure calculateMetrics runs synchronously without await calls to maintain sub-second latency.

Official References