Redis Beyond Caching: Pub/Sub, Sorted Sets, and Rate Limiting

#redis #backend #node #webdev

Redis Beyond Caching: Pub/Sub, Sorted Sets, and Rate Limiting

Most developers use Redis for caching. That's 20% of what it can do. Here are the patterns that make Redis indispensable.

Pub/Sub for Real-Time

import { createClient } from 'redis';

const publisher = createClient();
const subscriber = createClient();

await publisher.connect();
await subscriber.connect();

// Subscribe to a channel
await subscriber.subscribe('notifications', (message) => {
  const payload = JSON.parse(message);
  sendToWebSocket(payload.userId, payload);
});

// Publish from anywhere in your app
await publisher.publish('notifications', JSON.stringify({
  userId: '123',
  type: 'order_shipped',
  orderId: '456',
}));

Sorted Sets: Leaderboards and Rankings

// Add users to leaderboard with their score
await redis.zAdd('leaderboard', [
  { score: 2500, value: 'user:alice' },
  { score: 1800, value: 'user:bob' },
  { score: 3200, value: 'user:carol' },
]);

// Get top 10 with scores
const top10 = await redis.zRangeWithScores('leaderboard', 0, 9, { REV: true });

// Get user's rank (0-indexed)
const rank = await redis.zRevRank('leaderboard', 'user:alice');
console.log(`Alice is ranked #${rank + 1}`);

// Increment score
await redis.zIncrBy('leaderboard', 100, 'user:alice');

Rate Limiting With Sliding Window

async function isRateLimited(key: string, limit: number, windowMs: number): Promise<boolean> {
  const now = Date.now();
  const windowStart = now - windowMs;

  const pipeline = redis.multi();
  pipeline.zRemRangeByScore(key, 0, windowStart); // Remove old entries
  pipeline.zAdd(key, [{ score: now, value: `${now}-${Math.random()}` }]);
  pipeline.zCard(key); // Count entries in window
  pipeline.expire(key, Math.ceil(windowMs / 1000));

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

  return count > limit;
}

// Usage
const limited = await isRateLimited(`rate:${userId}`, 100, 60000);
if (limited) return res.status(429).json({ error: 'Rate limit exceeded' });

Distributed Locks

async function withLock<T>(lockKey: string, ttlMs: number, fn: () => Promise<T>): Promise<T> {
  const lockValue = crypto.randomUUID();

  // Acquire lock (SET NX = only set if not exists)
  const acquired = await redis.set(lockKey, lockValue, {
    NX: true,
    PX: ttlMs,
  });

  if (!acquired) throw new Error('Could not acquire lock');

  try {
    return await fn();
  } finally {
    // Only release if we still own it
    const current = await redis.get(lockKey);
    if (current === lockValue) await redis.del(lockKey);
  }
}

// Prevent double-processing a webhook
await withLock(`webhook:${eventId}`, 30000, () => processWebhook(event));

Session Storage

// Store session with 24h TTL
await redis.setEx(
  `session:${sessionId}`,
  86400,
  JSON.stringify({ userId, role, createdAt: Date.now() })
);

// Read session
const session = JSON.parse(await redis.get(`session:${sessionId}`) ?? 'null');