🚀 BD08 — Redis Avancé & Patterns

1. Cache-Aside Pattern

L'application consulte Redis d'abord, puis la base si absent (cache miss), puis stocke le résultat dans Redis.

const redis = require('./redis');
const pool  = require('./db');

async function getProductById(id) {
  const key = `cache:product:${id}`;

  // 1. Vérifier le cache
  const cached = await redis.get(key);
  if (cached) {
    return JSON.parse(cached);  // Cache HIT
  }

  // 2. Cache MISS — requête DB
  const { rows } = await pool.query(
    'SELECT * FROM products WHERE id = $1', [id]
  );
  const product = rows[0];
  if (!product) return null;

  // 3. Stocker en cache (5 minutes)
  await redis.set(key, JSON.stringify(product), 'EX', 300);
  return product;
}

// Invalidation du cache à la mise à jour
async function updateProduct(id, data) {
  await pool.query('UPDATE products SET name=$1, price=$2 WHERE id=$3',
    [data.name, data.price, id]);
  await redis.del(`cache:product:${id}`);         // invalider
  await redis.del('cache:products:list');          // invalider la liste aussi
}

// Cache avec stampede protection (mutex)
async function getCachedData(key, fetchFn, ttl = 300) {
  const cached = await redis.get(key);
  if (cached) return JSON.parse(cached);

  const lockKey = `lock:${key}`;
  const locked = await redis.set(lockKey, '1', 'EX', 10, 'NX');
  if (!locked) {
    await new Promise(r => setTimeout(r, 100));
    return getCachedData(key, fetchFn, ttl);
  }
  try {
    const data = await fetchFn();
    await redis.set(key, JSON.stringify(data), 'EX', ttl);
    return data;
  } finally {
    await redis.del(lockKey);
  }
}

2. Sessions Express avec Redis

const express      = require('express');
const session      = require('express-session');
const connectRedis = require('connect-redis');
const Redis        = require('ioredis');

const app = express();
const redisClient = new Redis(process.env.REDIS_URL);
const RedisStore = connectRedis(session);

app.use(session({
  store: new RedisStore({ client: redisClient, prefix: 'sess:' }),
  secret: process.env.SESSION_SECRET,
  resave: false,
  saveUninitialized: false,
  cookie: {
    secure: process.env.NODE_ENV === 'production',
    httpOnly: true,
    maxAge: 24 * 60 * 60 * 1000  // 24h
  }
}));

// Utilisation
app.post('/login', async (req, res) => {
  const user = await authenticateUser(req.body.email, req.body.password);
  req.session.userId = user.id;
  req.session.role   = user.role;
  res.json({ message: 'Connecté' });
});

app.get('/profile', (req, res) => {
  if (!req.session.userId) return res.status(401).json({ error: 'Non autorisé' });
  res.json({ userId: req.session.userId });
});

3. Pub/Sub

const Redis = require('ioredis');
const publisher  = new Redis(process.env.REDIS_URL);
const subscriber = new Redis(process.env.REDIS_URL);

// Abonnement
await subscriber.subscribe('orders:new', 'payments:completed');

subscriber.on('message', (channel, message) => {
  const data = JSON.parse(message);
  if (channel === 'orders:new') handleNewOrder(data);
  if (channel === 'payments:completed') handlePayment(data);
});

// Publication
await publisher.publish('orders:new', JSON.stringify({
  orderId: 42, userId: 1, total: 99.99
}));

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

4. Redis Streams

Streams : journal append-only persistant, consommable par plusieurs consumer groups.

// XADD — ajouter au stream
const id = await redis.xadd('events:orders', '*',
  'orderId', '42',
  'userId',  '1',
  'total',   '99.99',
  'status',  'created'
);

// XREAD — lire depuis un offset
const results = await redis.xread('COUNT', 10, 'STREAMS', 'events:orders', '$');

// Consumer Groups
await redis.xgroup('CREATE', 'events:orders', 'workers', '$', 'MKSTREAM');

// Lire en tant que worker
const msgs = await redis.xreadgroup(
  'GROUP', 'workers', 'worker-1',
  'COUNT', 5, 'STREAMS', 'events:orders', '>'
);

// Acquitter un message traité
await redis.xack('events:orders', 'workers', messageId);

// Longueur du stream
const len = await redis.xlen('events:orders');

// Lire l'historique
const history = await redis.xrange('events:orders', '-', '+', 'COUNT', 100);

5. Verrous distribués (Redlock)

// npm install redlock
const Redlock = require('redlock');
const redlock = new Redlock([redis], {
  retryCount: 10,
  retryDelay: 200,
  retryJitter: 50
});

async function processPayment(paymentId) {
  const resource = `lock:payment:${paymentId}`;
  const ttl = 10000; // 10 secondes

  let lock;
  try {
    lock = await redlock.acquire([resource], ttl);
    // Traitement critique — une seule instance à la fois
    await doProcessPayment(paymentId);
  } catch (err) {
    if (err.name === 'LockError') {
      console.warn('Paiement déjà en cours de traitement');
    } else {
      throw err;
    }
  } finally {
    if (lock) await lock.release();
  }
}

// Rate limiting avec INCR
async function rateLimit(ip) {
  const key = `ratelimit:${ip}:${Math.floor(Date.now() / 60000)}`; // par minute
  const count = await redis.incr(key);
  if (count === 1) await redis.expire(key, 60);
  if (count > 100) throw new Error('Rate limit exceeded');
}

6. Pipeline & MULTI/EXEC

// Pipeline — grouper les commandes (1 round-trip)
const pipeline = redis.pipeline();
pipeline.set('key1', 'val1');
pipeline.incr('counter');
pipeline.hset('hash1', 'field1', 'value1');
pipeline.expire('key1', 300);
const results = await pipeline.exec();
// [[null, 'OK'], [null, 1], [null, 1], [null, 1]]

// MULTI/EXEC — transaction atomique Redis
const multi = redis.multi();
multi.decr('stock:p1');
multi.zadd('activity:log', Date.now(), 'purchase:p1');
multi.expire('stock:p1', 86400);
const [decrResult, zaddResult] = await multi.exec();

7. Scripts Lua (EVAL)

Les scripts Lua s'exécutent atomiquement côté serveur, sans round-trips intermédiaires.

// Incrémenter avec limite (rate limiter atomique)
const rateLimitScript = `
  local current = redis.call('INCR', KEYS[1])
  if current == 1 then
    redis.call('EXPIRE', KEYS[1], ARGV[1])
  end
  if current > tonumber(ARGV[2]) then
    return 0
  end
  return 1
`;

const allowed = await redis.eval(rateLimitScript, 1,
  `ratelimit:${ip}`,  // KEYS[1]
  '60',               // ARGV[1] — TTL
  '100'               // ARGV[2] — max requêtes
);

// Charger un script une fois (SCRIPT LOAD)
const sha = await redis.script('load', rateLimitScript);
const result = await redis.evalsha(sha, 1, `ratelimit:${ip}`, '60', '100');