Caching, Pub/Sub & Patterns

Caching Patterns

// Cache-aside (lazy loading) — most common
async function getUser(id) {
  const cacheKey = `user:${id}`;
  const cached = await redis.get(cacheKey);
  if (cached) return JSON.parse(cached);

  const user = await db.findUser(id);
  await redis.set(cacheKey, JSON.stringify(user), 'EX', 300);  // 5 min TTL
  return user;
}

// Write-through — update cache on every write
async function updateUser(id, data) {
  const user = await db.updateUser(id, data);
  await redis.set(`user:${id}`, JSON.stringify(user), 'EX', 300);
  return user;
}

// Cache invalidation on write
async function deleteUser(id) {
  await db.deleteUser(id);
  await redis.del(`user:${id}`);
}

// Hash-based caching for partial updates
async function updateUserField(id, field, value) {
  await db.updateUser(id, { [field]: value });
  await redis.hset(`user:${id}`, field, value);
}

// Multi-key invalidation pattern (tag-based)
await redis.sadd(`tag:user:${userId}:keys`, `posts:${userId}`, `activity:${userId}`);
// On user update, get all related keys and delete them
const keys = await redis.smembers(`tag:user:${userId}:keys`);
if (keys.length) await redis.del(...keys);

Rate Limiting

// Fixed window rate limiter
async function checkRateLimit(ip, limit = 100, windowSec = 60) {
  const key = `rate:${ip}:${Math.floor(Date.now() / (windowSec * 1000))}`;
  const count = await redis.incr(key);
  if (count === 1) await redis.expire(key, windowSec);
  return count <= limit;
}

// Sliding window with sorted set (more accurate)
async function slidingWindowLimit(ip, limit = 100, windowMs = 60000) {
  const key = `rate:sliding:${ip}`;
  const now = Date.now();
  const windowStart = now - windowMs;
  const requestId = `${now}-${Math.random()}`;

  const pipe = redis.pipeline();
  pipe.zadd(key, now, requestId);
  pipe.zremrangebyscore(key, 0, windowStart);   // remove old requests
  pipe.zcard(key);
  pipe.expire(key, Math.ceil(windowMs / 1000));
  const results = await pipe.exec();

  const count = results[2][1];
  return count <= limit;
}

Pub/Sub

// Publisher (ioredis)
import Redis from 'ioredis';
const pub = new Redis();

await pub.publish('notifications', JSON.stringify({
  type: 'NEW_MESSAGE',
  userId: 42,
  text: 'Hello!',
}));

// Subscriber (separate connection required!)
const sub = new Redis();
sub.subscribe('notifications', 'alerts');

sub.on('message', (channel, message) => {
  const data = JSON.parse(message);
  console.log(`[channel:${channel}]`, data);
});

// Pattern subscribe
sub.psubscribe('user:*:events');
sub.on('pmessage', (pattern, channel, message) => {
  console.log(pattern, channel, message);
});

// Note: sub connection can ONLY subscribe — cannot run other commands
// Use a separate connection for regular commands

Distributed Lock

// Simple distributed lock with SET NX
async function acquireLock(resource, ttlMs = 5000) {
  const lockKey = `lock:${resource}`;
  const lockValue = crypto.randomUUID();  // unique ID for this lock holder
  const result = await redis.set(lockKey, lockValue, 'PX', ttlMs, 'NX');
  return result === 'OK' ? lockValue : null;
}

async function releaseLock(resource, lockValue) {
  // Atomic check-and-delete using Lua to prevent releasing someone else's lock
  const script = `
    if redis.call("get", KEYS[1]) == ARGV[1] then
      return redis.call("del", KEYS[1])
    else
      return 0
    end
  `;
  return redis.eval(script, 1, `lock:${resource}`, lockValue);
}

// Usage
const lockValue = await acquireLock('send-email', 10000);
if (!lockValue) {
  throw new Error('Could not acquire lock');
}
try {
  await sendEmail();
} finally {
  await releaseLock('send-email', lockValue);
}

Transactions & Pipelining

// Pipeline — batch commands (no atomicity guarantee, but fewer round trips)
const pipe = redis.pipeline();
pipe.set('k1', 'v1');
pipe.get('k1');
pipe.incr('counter');
const results = await pipe.exec();  // [[null, 'OK'], [null, 'v1'], [null, 1]]

// Transaction (MULTI/EXEC — atomic, all or nothing)
const results = await redis
  .multi()
  .set('balance:alice', 900)
  .set('balance:bob', 1100)
  .exec();  // either both succeed or both fail

// WATCH — optimistic locking
async function transfer(from, to, amount) {
  await redis.watch(`balance:${from}`);
  const fromBalance = parseInt(await redis.get(`balance:${from}`));
  if (fromBalance < amount) {
    await redis.unwatch();
    throw new Error('Insufficient funds');
  }
  const result = await redis
    .multi()
    .decrby(`balance:${from}`, amount)
    .incrby(`balance:${to}`, amount)
    .exec();  // returns null if WATCH detected a change
  if (!result) throw new Error('Transaction aborted (concurrent modification)');
}