Kafka – Message Broker cho hệ thống phân tán với NodeJS

// src/kafka/kafka.service.ts
import { Injectable, OnModuleInit, OnModuleDestroy, Logger } from '@nestjs/common';
import { Kafka, Producer, Consumer, Admin } from 'kafkajs';

@Injectable()
export class KafkaService implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(KafkaService.name);
  private kafka: Kafka;
  private producer: Producer;
  private consumers: Map<string, Consumer> = new Map();

  constructor() {
    this.kafka = new Kafka({
      clientId: 'shopxyz-app',
      brokers: (process.env.KAFKA_BROKERS ?? 'localhost:9092').split(','),
      retry: { initialRetryTime: 100, retries: 8 },
    });
    this.producer = this.kafka.producer({
      allowAutoTopicCreation: true,
    });
  }

  async onModuleInit() {
    await this.producer.connect();
    this.logger.log('Kafka producer connected');
  }

  async onModuleDestroy() {
    await this.producer.disconnect();
    for (const consumer of this.consumers.values()) {
      await consumer.disconnect();
    }
  }

  // Gửi message
  async send(topic: string, messages: Array<{ key?: string; value: any }>): Promise<void> {
    await this.producer.send({
      topic,
      messages: messages.map(m => ({
        key: m.key,
        value: JSON.stringify(m.value),
      })),
    });
  }

  // Subscribe và consume messages
  async subscribe(
    topic: string,
    groupId: string,
    handler: (message: any, key?: string) => Promise<void>,
  ): Promise<void> {
    const consumer = this.kafka.consumer({ groupId });
    await consumer.connect();
    await consumer.subscribe({ topic, fromBeginning: false });

    await consumer.run({
      eachMessage: async ({ message, partition, offset }) => {
        try {
          const value = JSON.parse(message.value!.toString());
          const key = message.key?.toString();
          await handler(value, key);
          this.logger.debug(`Processed: topic=${topic} partition=${partition} offset=${offset}`);
        } catch (error) {
          this.logger.error(`Failed to process message: ${error.message}`);
          // Gửi vào DLQ (Dead Letter Queue)
          await this.send(`${topic}.DLQ`, [{
            value: { error: error.message, originalMessage: message.value!.toString() },
          }]);
        }
      },
    });

    this.consumers.set(`${groupId}-${topic}`, consumer);
  }
}

// src/events/kafka-events.service.ts
export const TOPICS = {
  USER_CREATED: 'user.created',
  ORDER_PLACED: 'order.placed',
  ORDER_SHIPPED: 'order.shipped',
  PAYMENT_COMPLETED: 'payment.completed',
  INVENTORY_UPDATED: 'inventory.updated',
} as const;

@Injectable()
export class KafkaEventsService {
  constructor(private kafkaService: KafkaService) {}

  async publishUserCreated(user: { id: string; email: string; name: string }) {
    await this.kafkaService.send(TOPICS.USER_CREATED, [{
      key: user.id,          // Key quyết định partition → cùng user luôn cùng partition → đảm bảo thứ tự
      value: {
        userId: user.id,
        email: user.email,
        name: user.name,
        timestamp: new Date().toISOString(),
      },
    }]);
  }

  async publishOrderPlaced(order: {
    orderId: string;
    userId: string;
    items: any[];
    total: number;
  }) {
    await this.kafkaService.send(TOPICS.ORDER_PLACED, [{
      key: order.userId,
      value: { ...order, timestamp: new Date().toISOString() },
    }]);
  }
}

// src/notifications/notification-consumer.service.ts
@Injectable()
export class NotificationConsumerService implements OnModuleInit {
  constructor(
    private kafkaService: KafkaService,
    private mailService: MailService,
  ) {}

  async onModuleInit() {
    // Consumer group 'notifications' nhận order events
    await this.kafkaService.subscribe(
      TOPICS.ORDER_PLACED,
      'notifications-service',
      async (message) => {
        await this.mailService.sendOrderConfirmation(message.userId, message.orderId);
      },
    );

    await this.kafkaService.subscribe(
      TOPICS.USER_CREATED,
      'notifications-service',
      async (message) => {
        await this.mailService.sendWelcomeEmail(message.email, message.name);
      },
    );
  }
}

// src/inventory/inventory-consumer.service.ts
@Injectable()
export class InventoryConsumerService implements OnModuleInit {
  async onModuleInit() {
    // Consumer group khác → nhận cùng message độc lập
    await this.kafkaService.subscribe(
      TOPICS.ORDER_PLACED,
      'inventory-service',  // Khác group → mỗi group đều nhận message
      async (message) => {
        for (const item of message.items) {
          await this.stockService.decrement(item.productId, item.quantity);
        }
      },
    );
  }
}

// main.ts — Microservice mode
const app = await NestFactory.createMicroservice(AppModule, {
  transport: Transport.KAFKA,
  options: {
    client: {
      brokers: ['localhost:9092'],
    },
    consumer: {
      groupId: 'my-group',
    },
  },
});

// Controller
@Controller()
export class EventsController {
  @MessagePattern('user.created')
  handleUserCreated(@Payload() data: any) {
    console.log('User created:', data);
  }

  @EventPattern('order.placed')
  handleOrderPlaced(@Payload() data: any) {
    console.log('Order placed:', data);
  }
}