Redis Advanced Patterns

Pub/Sub Messaging

import { createClient } from 'redis';

// Publisher
const publisher = createClient();
await publisher.connect();

async function publishEvent(channel: string, event: object): Promise<void> {
  await publisher.publish(channel, JSON.stringify(event));
}

// Subscriber
const subscriber = createClient();
await subscriber.connect();

await subscriber.subscribe('user-events', (message) => {
  const event = JSON.parse(message);
  console.log('Received:', event);
});

await subscriber.subscribe('order-events', (message) => {
  handleOrderEvent(JSON.parse(message));
});

// Publish
await publishEvent('user-events', { type: 'user.created', userId: '123' });

Redis Streams (Persistent Pub/Sub)

// Producer: add to stream
await client.xAdd('events', '*', {
  type: 'order.placed',
  orderId: 'order_123',
  userId: 'user_456',
  amount: '99.99',
  timestamp: Date.now().toString(),
});

// Consumer group setup
await client.xGroupCreate('events', 'order-processor', '
#39;, { MKSTREAM: true });

// Consumer: read from group
while (true) {
  const messages = await client.xReadGroup(
    'order-processor',
    'consumer-1',
    [{ key: 'events', id: '>' }],
    { COUNT: 10, BLOCK: 2000 }
  );

  if (messages) {
    for (const { name, messages: msgs } of messages) {
      for (const { id, message } of msgs) {
        await processEvent(message);
        // Acknowledge processed message
        await client.xAck('events', 'order-processor', id);
      }
    }
  }
}

Rate Limiting with Sliding Window

async function checkRateLimit(
  userId: string,
  limit: number,
  windowMs: number
): Promise<{ allowed: boolean; remaining: number; resetIn: number }> {
  const key = `rate:${userId}`;
  const now = Date.now();
  const windowStart = now - windowMs;

  // Atomic operation with pipeline
  const pipeline = client.multi();
  pipeline.zRemRangeByScore(key, '-inf', windowStart.toString());  // Remove old requests
  pipeline.zAdd(key, { score: now, value: now.toString() });        // Add current request
  pipeline.zCard(key);                                               // Count requests in window
  pipeline.expire(key, Math.ceil(windowMs / 1000));                  // Set TTL

  const results = await pipeline.exec();
  const count = results[2] as number;

  return {
    allowed: count <= limit,
    remaining: Math.max(0, limit - count),
    resetIn: windowMs,
  };
}

// Middleware
app.use(async (req, res, next) => {
  const result = await checkRateLimit(req.ip, 100, 60000);
  res.setHeader('X-RateLimit-Limit', 100);
  res.setHeader('X-RateLimit-Remaining', result.remaining);

  if (!result.allowed) {
    return res.status(429).json({ error: 'Too many requests' });
  }
  next();
});

Distributed Lock

class RedisLock {
  constructor(private client: RedisClient) {}

  async acquire(resource: string, ttlMs: number): Promise<string | null> {
    const token = crypto.randomUUID();
    const key = `lock:${resource}`;

    const acquired = await this.client.set(key, token, {
      NX: true,          // Only set if not exists
      PX: ttlMs,         // Expire in ttlMs
    });

    return acquired ? token : null;
  }

  async release(resource: string, token: string): Promise<boolean> {
    // Atomic check-and-delete with Lua
    const script = `
      if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('del', KEYS[1])
      else
        return 0
      end
    `;
    const result = await this.client.eval(script, { keys: [`lock:${resource}`], arguments: [token] });
    return result === 1;
  }
}

// Usage
const lock = new RedisLock(client);

async function processOrder(orderId: string): Promise<void> {
  const token = await lock.acquire(`order:${orderId}`, 30000);
  if (!token) {
    throw new Error('Order is already being processed');
  }

  try {
    await doProcessOrder(orderId);
  } finally {
    await lock.release(`order:${orderId}`, token);
  }
}

Sorted Sets for Leaderboards

// Add/update score
await client.zAdd('game:leaderboard', { score: 1500, value: 'player:alice' });

// Get top 10
const top10 = await client.zRangeWithScores('game:leaderboard', 0, 9, { REV: true });
// Returns: [{ value: 'player:alice', score: 1500 }, ...]

// Get player rank (0-indexed)
const rank = await client.zRevRank('game:leaderboard', 'player:alice');

// Get players in score range
const midLevel = await client.zRangeByScoreWithScores('game:leaderboard', '1000', '2000');

Redis's data structures enable elegant solutions to complex real-time problems.