Redis Client

The entrolytics/redis-client package provides a Redis client for real-time analytics, caching, and stream processing with Entrolytics integration.

Installation

npm install @entrolytics/redis-client

Quick Start

Basic Redis Integration

import { EntrolyticsRedisClient } from '@entrolytics/redis-client'
import Redis from 'ioredis'

// Initialize Redis client
const redis = new Redis({
  host: 'localhost',
  port: 6379,
  password: 'your-redis-password'
})

// Initialize Entrolytics Redis client
const analytics = new EntrolyticsRedisClient({
  redis,
  websiteId: 'your-website-id',
  apiKey: 'your-api-key'
})

// Track events with Redis caching
await analytics.track('user_action', {
  userId: 'user-123',
  action: 'purchase',
  amount: 99.99
})

// Real-time metrics
const metrics = await analytics.getRealtimeMetrics('user_actions', {
  timeRange: '1h',
  groupBy: 'action'
})

console.log('Real-time metrics:', metrics)

Stream Processing

import { EntrolyticsRedisClient } from '@entrolytics/redis-client'

// Set up stream processing
await analytics.setupStream('events', {
  maxLen: 10000,
  consumerGroup: 'analytics-processors'
})

// Process events in real-time
analytics.processStream('events', async (event) => {
  console.log('Processing event:', event)

  // Enrich event data
  const enrichedEvent = await enrichEventData(event)

  // Send to Entrolytics
  await analytics.track(enrichedEvent.event, enrichedEvent.properties)

  // Update Redis metrics
  await analytics.updateMetrics(enrichedEvent)
})

Caching Layer

// Use Redis as caching layer for analytics
const getCachedAnalytics = async (key, ttl = 300) => {
  const cached = await redis.get(`analytics:${key}`)

  if (cached) {
    return JSON.parse(cached)
  }

  const data = await analytics.fetchAnalytics(key)
  await redis.setex(`analytics:${key}`, ttl, JSON.stringify(data))

  return data
}

// Usage
const dashboardData = await getCachedAnalytics('dashboard:overview', 60)

Configuration

Client Options

const analytics = new EntrolyticsRedisClient({
  redis: redisInstance, // Redis client instance
  websiteId: 'your-website-id', // required
  apiKey: 'your-api-key', // optional
  host: 'https://entrolytics.dev', // custom host
  streamKey: 'entrolytics:events', // Redis stream key
  metricsKey: 'entrolytics:metrics', // Redis metrics key
  cacheKey: 'entrolytics:cache', // Redis cache key prefix
  streamMaxLen: 10000, // Max stream length
  batchSize: 100, // Batch size for processing
  flushInterval: 5000, // Flush interval (ms)
  retryAttempts: 3, // Retry attempts
  retryDelay: 1000, // Retry delay (ms)
  debug: false // Debug mode
})

Environment Variables

# .env
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_PASSWORD=your-redis-password
ENTROLYTICS_WEBSITE_ID=your-website-id
ENTROLYTICS_API_KEY=your-api-key
ENTROLYTICS_REDIS_STREAM_KEY=entrolytics:events
ENTROLYTICS_REDIS_METRICS_KEY=entrolytics:metrics
ENTROLYTICS_REDIS_CACHE_KEY=entrolytics:cache

Redis Configuration

// Redis client configuration
const redis = new Redis({
  host: process.env.REDIS_HOST || 'localhost',
  port: parseInt(process.env.REDIS_PORT) || 6379,
  password: process.env.REDIS_PASSWORD,
  db: 0,
  retryDelayOnFailover: 100,
  maxRetriesPerRequest: 3,
  lazyConnect: true,
  keepAlive: 30000,
  connectTimeout: 10000,
  commandTimeout: 5000
})

API Reference

Core Client

class EntrolyticsRedisClient {
  constructor(options)

  // Event tracking with Redis
  async track(event, properties = {})
  async trackBatch(events)
  async page(url, properties = {})
  async identify(userId, traits = {})

  // Stream operations
  async setupStream(streamName, options = {})
  async addToStream(streamName, event)
  async processStream(streamName, processor, options = {})
  async getStreamLength(streamName)

  // Metrics operations
  async incrementMetric(metric, value = 1, labels = {})
  async getMetric(metric, labels = {})
  async getMetrics(pattern, timeRange = null)
  async setMetric(metric, value, labels = {})
  async deleteMetric(metric, labels = {})

  // Cache operations
  async get(key)
  async set(key, value, ttl = null)
  async setex(key, ttl, value)
  async del(key)
  async exists(key)

  // Real-time analytics
  async getRealtimeMetrics(metric, options = {})
  async getTopEvents(timeRange = '1h', limit = 10)
  async getActiveUsers(timeRange = '1h')
  async getEventCounts(events, timeRange = '1h')

  // Aggregation
  async aggregateEvents(timeRange, groupBy, filters = {})
  async aggregateMetrics(timeRange, metrics, filters = {})

  // Cleanup
  async flush()
  async cleanup(ttl = 86400)
}

Stream Processing

// Stream processor options
const streamOptions = {
  consumerGroup: 'analytics-processors',
  consumerName: 'processor-1',
  batchSize: 100,
  blockTime: 5000,
  maxRetries: 3,
  retryDelay: 1000,
  deadLetterQueue: 'entrolytics:dlq'
}

// Event processor
const processor = async (event) => {
  console.log('Processing:', event)

  // Custom processing logic
  const processed = await processEvent(event)

  // Update metrics
  await analytics.incrementMetric('events_processed', 1, {
    event_type: event.event,
    processor: 'main'
  })

  return processed
}

// Start processing
await analytics.processStream('events', processor, streamOptions)

Metrics API

// Increment counters
await analytics.incrementMetric('page_views', 1, {
  page: '/home',
  user_type: 'authenticated'
})

// Set gauge values
await analytics.setMetric('active_users', 1250, {
  region: 'us-east-1'
})

// Get metrics with time range
const metrics = await analytics.getMetrics('page_views:*', {
  timeRange: '1h',
  aggregation: 'sum'
})

// Real-time metrics
const realtime = await analytics.getRealtimeMetrics('user_actions', {
  timeRange: '5m',
  groupBy: 'action',
  filters: { user_type: 'premium' }
})

Advanced Usage

Real-time Dashboard

// Real-time dashboard data provider
class RealtimeDashboard {
  constructor(analytics) {
    this.analytics = analytics
  }

  async getOverview(timeRange = '1h') {
    const [totalEvents, activeUsers, topPages, topEvents, errorRate] =
      await Promise.all([
        this.analytics.getMetric('total_events', { timeRange }),
        this.analytics.getActiveUsers(timeRange),
        this.analytics.getTopPages(timeRange, 10),
        this.analytics.getTopEvents(timeRange, 10),
        this.analytics.getMetric('error_rate', { timeRange })
      ])

    return {
      totalEvents,
      activeUsers,
      topPages,
      topEvents,
      errorRate
    }
  }

  async getEventTimeline(event, timeRange = '1h') {
    return await this.analytics.getMetrics(`event:${event}:timeline`, {
      timeRange,
      granularity: '1m'
    })
  }

  async getUserActivity(userId, timeRange = '24h') {
    return await this.analytics.getMetrics(`user:${userId}:*`, {
      timeRange,
      groupBy: 'action'
    })
  }
}

// Usage
const dashboard = new RealtimeDashboard(analytics)

// WebSocket endpoint for real-time updates
app.ws('/dashboard/realtime', (ws) => {
  const updateDashboard = async () => {
    const data = await dashboard.getOverview('5m')
    ws.send(JSON.stringify(data))
  }

  const interval = setInterval(updateDashboard, 5000)

  ws.on('close', () => clearInterval(interval))
})

Event Enrichment

// Event enrichment pipeline
class EventEnricher {
  constructor(analytics, redis) {
    this.analytics = analytics
    this.redis = redis
  }

  async enrichEvent(event) {
    const enriched = { ...event }

    // Add user context
    if (event.userId) {
      enriched.userContext = await this.getUserContext(event.userId)
    }

    // Add session context
    if (event.sessionId) {
      enriched.sessionContext = await this.getSessionContext(event.sessionId)
    }

    // Add geographic context
    if (event.ip) {
      enriched.geoContext = await this.getGeoContext(event.ip)
    }

    // Add device context
    if (event.userAgent) {
      enriched.deviceContext = await this.getDeviceContext(event.userAgent)
    }

    return enriched
  }

  async getUserContext(userId) {
    const cacheKey = `user:${userId}:context`
    let context = await this.redis.get(cacheKey)

    if (!context) {
      context = await this.fetchUserContext(userId)
      await this.redis.setex(cacheKey, 300, JSON.stringify(context))
    }

    return JSON.parse(context)
  }

  async getSessionContext(sessionId) {
    return await this.redis.hgetall(`session:${sessionId}`)
  }

  async getGeoContext(ip) {
    // Implement IP geolocation
    return {
      country: 'US',
      region: 'CA',
      city: 'San Francisco'
    }
  }

  async getDeviceContext(userAgent) {
    // Implement device detection
    return {
      type: 'desktop',
      os: 'macOS',
      browser: 'Chrome'
    }
  }
}

// Usage with stream processing
const enricher = new EventEnricher(analytics, redis)

analytics.processStream('events', async (event) => {
  const enriched = await enricher.enrichEvent(event)
  await analytics.track(enriched.event, enriched.properties)
})

Rate Limiting and Throttling

// Rate limiting with Redis
class RateLimiter {
  constructor(redis) {
    this.redis = redis
  }

  async isAllowed(key, limit, window) {
    const current = await this.redis.incr(key)

    if (current === 1) {
      await this.redis.expire(key, window)
    }

    return current <= limit
  }

  async getRemaining(key, limit, window) {
    const current = (await this.redis.get(key)) || 0
    return Math.max(0, limit - parseInt(current))
  }
}

// Apply rate limiting to analytics
const rateLimiter = new RateLimiter(redis)

const trackWithRateLimit = async (event, properties) => {
  const key = `rate_limit:analytics:${event}`
  const allowed = await rateLimiter.isAllowed(key, 100, 60) // 100 events per minute

  if (allowed) {
    await analytics.track(event, properties)
  } else {
    // Queue for later processing
    await analytics.addToStream('rate_limited', { event, properties })
  }
}

Background Processing

// Background event processor
class BackgroundProcessor {
  constructor(analytics, redis) {
    this.analytics = analytics
    this.redis = redis
    this.processing = false
  }

  async start() {
    this.processing = true

    while (this.processing) {
      try {
        const events = await this.redis.xreadgroup(
          'GROUP',
          'processors',
          'bg-processor',
          'COUNT',
          '100',
          'BLOCK',
          '5000',
          'STREAMS',
          'background_events',
          '>'
        )

        if (events && events.length > 0) {
          await this.processEvents(events)
        }
      } catch (error) {
        console.error('Background processing error:', error)
        await this.sleep(5000)
      }
    }
  }

  async processEvents(events) {
    const processed = []

    for (const [stream, streamEvents] of events) {
      for (const [id, event] of streamEvents) {
        try {
          await this.analytics.track(event.event, event.properties)
          processed.push([stream, id])
        } catch (error) {
          console.error('Failed to process event:', error)
        }
      }
    }

    // Acknowledge processed events
    if (processed.length > 0) {
      await this.redis.xack(
        'background_events',
        'processors',
        ...processed.map((p) => p[1])
      )
    }
  }

  stop() {
    this.processing = false
  }

  sleep(ms) {
    return new Promise((resolve) => setTimeout(resolve, ms))
  }
}

// Start background processor
const processor = new BackgroundProcessor(analytics, redis)
processor.start()

Testing

Unit Tests

import { EntrolyticsRedisClient } from '@entrolytics/redis-client'
import Redis from 'ioredis-mock'

describe('EntrolyticsRedisClient', () => {
  let analytics
  let mockRedis

  beforeEach(() => {
    mockRedis = new Redis()
    analytics = new EntrolyticsRedisClient({
      redis: mockRedis,
      websiteId: 'test-website-id'
    })
  })

  afterEach(async () => {
    await mockRedis.flushall()
  })

  test('should track events', async () => {
    await analytics.track('test_event', { property: 'value' })

    const events = await mockRedis.xrange('entrolytics:events', '-', '+')
    expect(events).toHaveLength(1)

    const event = JSON.parse(events[0][1].event)
    expect(event.event).toBe('test_event')
    expect(event.properties.property).toBe('value')
  })

  test('should increment metrics', async () => {
    await analytics.incrementMetric('test_metric', 5, { tag: 'test' })

    const value = await analytics.getMetric('test_metric', { tag: 'test' })
    expect(value).toBe(5)
  })

  test('should cache data', async () => {
    await analytics.set('test_key', { data: 'test' }, 300)

    const cached = await analytics.get('test_key')
    expect(cached.data).toBe('test')
  })
})

Integration Tests

describe('Redis Integration', () => {
  let analytics
  let redis

  beforeAll(async () => {
    redis = new Redis({
      host: 'localhost',
      port: 6379,
      db: 15 // Use test database
    })

    analytics = new EntrolyticsRedisClient({
      redis,
      websiteId: 'integration-test'
    })
  })

  afterAll(async () => {
    await redis.flushdb()
    await redis.quit()
  })

  test('should process events in real-time', async () => {
    const processedEvents = []

    await analytics.setupStream('test_events')

    analytics.processStream('test_events', async (event) => {
      processedEvents.push(event)
    })

    // Add events to stream
    await analytics.addToStream('test_events', {
      event: 'test_event',
      properties: { test: true }
    })

    // Wait for processing
    await new Promise((resolve) => setTimeout(resolve, 1000))

    expect(processedEvents).toHaveLength(1)
    expect(processedEvents[0].event).toBe('test_event')
  })
})

Performance Optimization

Connection Pooling

import Redis from 'ioredis'

// Configure connection pool
const redis = new Redis({
  host: 'localhost',
  port: 6379,
  maxRetriesPerRequest: 3,
  retryDelayOnFailover: 100,
  lazyConnect: true,
  keepAlive: 30000,
  family: 4,
  connectTimeout: 10000,
  commandTimeout: 5000
})

// Multiple Redis instances for different purposes
const analytics = new EntrolyticsRedisClient({
  redis: redis, // Main operations
  cacheRedis: new Redis({ db: 1 }), // Cache operations
  streamRedis: new Redis({ db: 2 }) // Stream operations
})

Batch Operations

// Batch event processing
class BatchProcessor {
  constructor(analytics, batchSize = 100, flushInterval = 5000) {
    this.analytics = analytics
    this.batchSize = batchSize
    this.flushInterval = flushInterval
    this.batch = []
    this.timer = null
  }

  add(event) {
    this.batch.push(event)

    if (this.batch.length >= this.batchSize) {
      this.flush()
    } else if (!this.timer) {
      this.timer = setTimeout(() => this.flush(), this.flushInterval)
    }
  }

  async flush() {
    if (this.batch.length === 0) return

    const events = this.batch.splice(0)
    this.timer = null

    await this.analytics.trackBatch(events)
  }
}

// Usage
const batchProcessor = new BatchProcessor(analytics)

// Add events to batch
batchProcessor.add({ event: 'user_action', properties: { type: 'click' } })
batchProcessor.add({ event: 'page_view', properties: { page: '/home' } })

Memory Management

// Memory-efficient stream processing
const processStreamEfficiently = async (streamName, processor) => {
  let lastId = '$'

  while (true) {
    try {
      const results = await redis.xread(
        'COUNT',
        '10',
        'BLOCK',
        '1000',
        'STREAMS',
        streamName,
        lastId
      )

      if (!results) continue

      const [[, events]] = results

      for (const [id, event] of events) {
        await processor(event)
        lastId = id
      }

      // Acknowledge processed events
      await redis.xack(streamName, 'processors', ...events.map((e) => e[0]))
    } catch (error) {
      console.error('Stream processing error:', error)
      await new Promise((resolve) => setTimeout(resolve, 5000))
    }
  }
}

Troubleshooting

Best Practices

Migration Guide

From Direct Redis Usage

// Before - Direct Redis
import Redis from 'ioredis'

const redis = new Redis()
await redis.set('analytics:event', JSON.stringify(event))

// After - Entrolytics Redis Client
import { EntrolyticsRedisClient } from '@entrolytics/redis-client'

const analytics = new EntrolyticsRedisClient({ redis, websiteId: 'your-id' })
await analytics.track(event.event, event.properties)

From Other Analytics Libraries

// Before - Other analytics
import Analytics from 'other-analytics'

Analytics.track('event', properties)

// After - Entrolytics with Redis
import { EntrolyticsRedisClient } from '@entrolytics/redis-client'

const analytics = new EntrolyticsRedisClient({ redis, websiteId: 'your-id' })
await analytics.track('event', properties)

// Additional Redis features
await analytics.incrementMetric('counter', 1)
const cached = await analytics.get('cache:key')

Redis client for Entrolytics - First-party growth analytics for the edge

Redis Client

Redis connection issues

Stream processing delays

Memory leaks

Performance

Reliability

Security

Data Management

On this page