Apache Kafka

1/*
2 * Architecture Kafka
3 *
4 * TOPICS — canaux de messages (comme une table de DB)
5 * user.events        → inscription, mise à jour profil, suppression
6 * tutorial.events    → création, publication, modification, suppression
7 * progress.events    → leçon commencée, complétée, tutoriel terminé
8 * payment.events     → paiement initié, succès, échec, remboursement
9 *
10 * PARTITIONS — parallélisme et scalabilité
11 * Un topic est divisé en partitions
12 * Partition 0 : users A-H
13 * Partition 1 : users I-P
14 * Partition 2 : users Q-Z
15 * → Les messages de la même partition sont ORDONNÉS
16 * → Différentes partitions → différents consumers (parallélisme)
17 *
18 * CONSUMER GROUPS — traitement distribué
19 * Groupe "notifications" : NotificationService consomme user.events
20 * Groupe "analytics"     : AnalyticsService consomme user.events
21 * → Chaque groupe reçoit TOUS les messages (broadcast)
22 * → Dans un groupe, chaque partition est lue par 1 seul consumer
23 *
24 * OFFSET — position dans la partition
25 * Chaque consumer garde sa position (offset)
26 * → Peut rejouer les messages en cas de bug
27 * → Les messages sont gardés N jours (rétention configurable)
28 */
29
30import { Kafka, Partitioners } from "kafkajs";
31
32const kafka = new Kafka({
33  clientId: "usher-api",
34  brokers: ["kafka-1:9092", "kafka-2:9092"],
35  retry: { initialRetryTime: 300, retries: 10 },
36});
37
38// Producteur — publier des événements
39const producer = kafka.producer({
40  createPartitioner: Partitioners.DefaultPartitioner,
41});
42
43await producer.connect();
44
45// Publier un événement après inscription
46async function publishUserRegistered(user: User): Promise<void> {
47  await producer.send({
48    topic: "user.events",
49    messages: [{
50      key: user.id,           // Clé de partition (même user → même partition → ordonné)
51      value: JSON.stringify({
52        type: "UserRegistered",
53        version: "1.0",
54        payload: {
55          userId: user.id,
56          email: user.email,
57          firstName: user.firstName,
58          registeredAt: new Date().toISOString(),
59        },
60        metadata: {
61          sourceService: "auth-service",
62          traceId: context.traceId,
63        },
64      }),
65      headers: { "content-type": "application/json" },
66    }],
67  });
68}
69
70// Consommateur — NotificationService
71const consumer = kafka.consumer({ groupId: "notification-service" });
72await consumer.connect();
73await consumer.subscribe({ topic: "user.events", fromBeginning: false });
74
75await consumer.run({
76  partitionsConsumedConcurrently: 3,
77  eachMessage: async ({ topic, partition, message }) => {
78    const event = JSON.parse(message.value!.toString());
79    switch (event.type) {
80      case "UserRegistered":
81        await sendWelcomeEmail(event.payload);
82        break;
83      case "UserDeleted":
84        await cleanupUserData(event.payload.userId);
85        break;
86    }
87  },
88});