Projet Final — Analytics Dashboard API

🎯 Contexte & Objectif

Vous êtes développeur backend dans une startup e-commerce. L'équipe a besoin d'un Analytics Dashboard API capable d'agréger des données de sources hétérogènes en temps réel.

Ce projet met en pratique tous les modules de la formation : PostgreSQL (transactions, full-text), MongoDB (agrégations, Change Streams), Redis (cache multi-couche, Pub/Sub, sessions), Prisma ORM, et les patterns d'architecture avancés.

🗄️ Architecture Polyglot Persistence

📡 Endpoints API

🔄 Flux de commande

// POST /orders — flux complet
async function createOrder(userId, items) {
  // 1. Vérifier stock dans MongoDB (lecture rapide)
  for (const item of items) {
    const product = await mongoDb.collection('products').findOne(
      { _id: item.productId, stock: { $gte: item.qty } }
    );
    if (!product) throw new Error(`Stock insuffisant: ${item.productId}`);
  }

  // 2. Créer la commande dans PostgreSQL (ACID)
  const order = await prisma.$transaction(async (tx) => {
    const o = await tx.order.create({
      data: { userId, status: 'confirmed',
        items: { createMany: { data: items } },
        total: items.reduce((s, i) => s + i.price * i.qty, 0) }
    });
    await tx.payment.create({ data: { orderId: o.id, status: 'pending' } });
    return o;
  });

  // 3. Décrémenter le stock dans MongoDB
  for (const item of items) {
    await mongoDb.collection('products').updateOne(
      { _id: item.productId },
      { $inc: { stock: -item.qty } }
    );
  }

  // 4. Publier event Redis pour projections
  await redis.publish('order:created', JSON.stringify({
    orderId: order.id, userId, total: order.total
  }));

  // 5. Mettre à jour le leaderboard Redis
  await redis.zincrby('leaderboard:revenue', order.total, `user:${userId}`);

  return order;
}

📡 Change Streams → Redis Pub/Sub

// Listener Change Streams MongoDB → Redis
async function startChangeStreamBridge(mongoDb, redis) {
  const stream = mongoDb.collection('orders').watch(
    [{ $match: { operationType: { $in: ['insert','update'] } } }],
    { fullDocument: 'updateLookup' }
  );

  stream.on('change', async (change) => {
    const doc = change.fullDocument;
    if (!doc) return;

    // Invalider le cache analytics
    await redis.del('cache:analytics:kpis');
    await redis.del('cache:analytics:products');

    // Publier pour SSE temps réel
    await redis.publish('realtime:kpis', JSON.stringify({
      event: change.operationType,
      orderId: doc._id,
      total: doc.total,
      ts: Date.now()
    }));
  });

  return stream;
}

⚡ Analytics en temps réel (SSE)

// GET /analytics/realtime — Server-Sent Events
app.get('/analytics/realtime', authMiddleware, (req, res) => {
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');
  res.flushHeaders();

  const sub = new Redis(process.env.REDIS_URL);
  sub.subscribe('realtime:kpis');

  sub.on('message', (channel, msg) => {
    res.write(`data: ${msg}\n\n`);
  });

  // Heartbeat toutes les 30s
  const heartbeat = setInterval(() => res.write(':ping\n\n'), 30000);

  req.on('close', () => {
    clearInterval(heartbeat);
    sub.unsubscribe();
    sub.quit();
  });
});

🗺️ Étapes de réalisation

🐳 Infrastructure Docker

# docker-compose.yml
version: '3.9'
services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: analytics
      POSTGRES_USER: admin
      POSTGRES_PASSWORD: secret
    ports: ["5432:5432"]
    volumes: [pg_data:/var/lib/postgresql/data]

  mongodb:
    image: mongo:7
    ports: ["27017:27017"]
    environment:
      MONGO_INITDB_ROOT_USERNAME: admin
      MONGO_INITDB_ROOT_PASSWORD: secret
    volumes: [mongo_data:/data/db]
    command: --replSet rs0  # pour Change Streams

  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]
    command: redis-server --save 60 1 --loglevel warning

volumes:
  pg_data:
  mongo_data:

✅ Critères de validation

• ☐ POST /orders : rollback PG si décrémentation Mongo échoue
• ☐ GET /products : cache Redis + header X-Cache: HIT/MISS
• ☐ GET /analytics/kpis : données fraîches en < 100ms (via cache)
• ☐ Change Stream invalide le cache à chaque nouvelle commande
• ☐ SSE /analytics/realtime reçoit les événements en temps réel
• ☐ Rate limiter bloque > 100 req/min par IP
• ☐ Leaderboard Redis mis à jour à chaque commande confirmée
• ☐ GET /health retourne latence des 3 bases (< 50ms chacune)
• ☐ Tests Jest : couverture ≥ 70% sur les services critiques
• ☐ Aucune requête SQL non paramétrée

🌟 Bonus

• Réplication PostgreSQL : pool read/write séparés avec fallback
• Distributed lock (Redlock) pour les commandes concurrentes
• Redis Streams pour l'event sourcing des commandes
• Prisma soft delete + audit log middleware
• Dashboard HTML simple consommant l'API (vanilla JS)