Redis Mastery: Caching Strategies, Pub/Sub, Streams & Avoiding Cache Stampedes

const redis = new Redis(process.env.REDIS_URL);

// Simple cache
await redis.set("user:123", JSON.stringify(user), "EX", 3600); // 1 hour TTL
const cached = await redis.get("user:123");

// Atomic counter
await redis.incr("page_views:post:456");
await redis.incrby("inventory:product:789", -1);

// Store user session data — no need to serialize/deserialize entire object
await redis.hset("session:abc123", {
  userId: "user-456",
  role: "admin",
  loginAt: Date.now().toString(),
  ipAddress: "203.0.113.1",
});

// Get specific field — more efficient than GET + parse
const role = await redis.hget("session:abc123", "role");

// Get all fields
const session = await redis.hgetall("session:abc123");

// Leaderboard — automatically sorted by score
await redis.zadd("leaderboard:monthly", 9840, "user:alice");
await redis.zadd("leaderboard:monthly", 8721, "user:bob");
await redis.zadd("leaderboard:monthly", 11203, "user:charlie");

// Top 10 users
const top10 = await redis.zrevrange("leaderboard:monthly", 0, 9, "WITHSCORES");

// User's rank
const rank = await redis.zrevrank("leaderboard:monthly", "user:alice");

async function getPost(postId: string): Promise<Post> {
  const cacheKey = `post:${postId}`;

  // 1. Check cache
  const cached = await redis.get(cacheKey);
  if (cached) {
    return JSON.parse(cached);
  }

  // 2. Cache miss — fetch from database
  const post = await db.query(
    "SELECT * FROM posts WHERE id = $1 AND status = 'published'",
    [postId]
  );

  if (!post) throw new Error("Post not found");

  // 3. Populate cache
  await redis.set(cacheKey, JSON.stringify(post), "EX", 3600);

  return post;
}

// Solution: Distributed lock — only one request rebuilds the cache
import { Mutex } from "async-mutex";

const mutexes = new Map<string, Mutex>();

async function getPostWithLock(postId: string): Promise<Post> {
  const cacheKey = `post:${postId}`;
  const lockKey = `lock:${cacheKey}`;

  // Try cache first (no lock needed for hits)
  const cached = await redis.get(cacheKey);
  if (cached) return JSON.parse(cached);

  // Acquire distributed lock using Redis SET NX
  const lockAcquired = await redis.set(lockKey, "1", "NX", "EX", 10);

  if (!lockAcquired) {
    // Another process is rebuilding — wait and retry
    await new Promise((r) => setTimeout(r, 100));
    const retried = await redis.get(cacheKey);
    if (retried) return JSON.parse(retried);
  }

  try {
    const post = await db.query("SELECT * FROM posts WHERE id = $1", [postId]);
    await redis.set(cacheKey, JSON.stringify(post), "EX", 3600);
    return post;
  } finally {
    await redis.del(lockKey); // Always release the lock
  }
}

// Alternative: Probabilistic Early Expiration (XFetch algorithm)
// Start refreshing the cache before it expires — probability increases as TTL decreases
async function getPostXFetch(postId: string): Promise<Post> {
  const cacheKey = `post:${postId}`;
  const result = await redis.get(cacheKey);
  
  if (result) {
    const { data, ttl, delta } = JSON.parse(result);
    const remainingTTL = await redis.ttl(cacheKey);
    
    // Probabilistically recompute before expiry
    const shouldRefresh = remainingTTL - delta * Math.log(Math.random()) < 0;
    
    if (!shouldRefresh) return data;
  }
  
  const start = Date.now();
  const post = await db.query("SELECT * FROM posts WHERE id = $1", [postId]);
  const delta = (Date.now() - start) / 1000; // Compute time in seconds
  
  await redis.set(
    cacheKey,
    JSON.stringify({ data: post, ttl: 3600, delta }),
    "EX", 3600
  );
  
  return post;
}

// Publisher (e.g., when a new comment is posted)
const publisher = new Redis(process.env.REDIS_URL);

async function addComment(postId: string, comment: Comment) {
  await db.query("INSERT INTO comments ...", [...]);

  // Notify all subscribers watching this post
  await publisher.publish(
    `post:${postId}:comments`,
    JSON.stringify({ type: "new_comment", comment })
  );
}

// Subscriber (WebSocket server)
const subscriber = new Redis(process.env.REDIS_URL);

await subscriber.subscribe("post:abc123:comments");

subscriber.on("message", (channel, message) => {
  const event = JSON.parse(message);
  // Broadcast to all WebSocket clients watching this post
  wsServer.to(channel).emit("comment", event.comment);
});

// Producer — add to stream
await redis.xadd(
  "user-events",
  "*",               // Auto-generate ID
  "userId", user.id,
  "event", "login",
  "ip", req.ip,
  "timestamp", Date.now().toString()
);

// Consumer group — multiple workers, each event processed once
await redis.xgroup("CREATE", "user-events", "analytics-workers", "$", "MKSTREAM");

// Worker
while (true) {
  const messages = await redis.xreadgroup(
    "GROUP", "analytics-workers", "worker-1",
    "COUNT", 10,
    "BLOCK", 2000,  // Block 2 seconds if no messages
    "STREAMS", "user-events", ">"
  );

  for (const [stream, entries] of messages ?? []) {
    for (const [id, fields] of entries) {
      await processEvent(Object.fromEntries(fields));
      await redis.xack("user-events", "analytics-workers", id); // Acknowledge
    }
  }
}