Kafka integration

Author: billsedison

.env.sample

@@ -1,6 +1,34 @@
 POSTGRES_SCHEMA="public"
 DATABASE_URL="postgresql://johndoe:randompassword@localhost:5432/mydb?schema=${POSTGRES_SCHEMA}"
 
+# Kafka Configuration
+KAFKA_BROKERS=localhost:9092
+KAFKA_CLIENT_ID=tc-review-api
+KAFKA_GROUP_ID=tc-review-consumer-group
+KAFKA_SSL_ENABLED=false
+
+# SASL Configuration (optional - uncomment if needed)
+# KAFKA_SASL_MECHANISM=plain
+# KAFKA_SASL_USERNAME=
+# KAFKA_SASL_PASSWORD=
+
+# Consumer Configuration
+KAFKA_SESSION_TIMEOUT=30000
+KAFKA_HEARTBEAT_INTERVAL=3000
+KAFKA_MAX_WAIT_TIME=5000
+KAFKA_CONNECTION_TIMEOUT=10000
+KAFKA_REQUEST_TIMEOUT=30000
+
+# Retry Configuration
+KAFKA_RETRY_ATTEMPTS=5
+KAFKA_INITIAL_RETRY_TIME=100
+KAFKA_MAX_RETRY_TIME=30000
+
+# Dead Letter Queue Configuration
+KAFKA_DLQ_ENABLED=true
+KAFKA_DLQ_TOPIC_SUFFIX=.dlq
+KAFKA_DLQ_MAX_RETRIES=3
+
 # API configs
 BUS_API_URL="https://api.topcoder-dev.com/v5/bus/events"
 CHALLENGE_API_URL="https://api.topcoder-dev.com/v5/challenges/"

docker-compose.kafka.yml

@@ -0,0 +1,54 @@
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.4.0
+    hostname: zookeeper
+    container_name: zookeeper
+    ports:
+      - "2181:2181"
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+    networks:
+      - kafka-network
+
+  kafka:
+    image: confluentinc/cp-kafka:7.4.0
+    hostname: kafka
+    container_name: kafka
+    depends_on:
+      - zookeeper
+    ports:
+      - "9092:9092"
+      - "9101:9101"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_JMX_PORT: 9101
+      KAFKA_JMX_HOSTNAME: localhost
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
+    networks:
+      - kafka-network
+
+  kafka-ui:
+    image: provectuslabs/kafka-ui:latest
+    container_name: kafka-ui
+    depends_on:
+      - kafka
+    ports:
+      - "8080:8080"
+    environment:
+      KAFKA_CLUSTERS_0_NAME: local
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
+      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181
+    networks:
+      - kafka-network
+
+networks:
+  kafka-network:
+    driver: bridge

package.json

@@ -39,7 +39,8 @@
     "nanoid": "~5.1.2",
     "reflect-metadata": "^0.2.2",
     "rxjs": "^7.8.1",
-    "tc-core-library-js": "appirio-tech/tc-core-library-js.git#v3.0.1"
+    "tc-core-library-js": "appirio-tech/tc-core-library-js.git#v3.0.1",
+    "kafkajs": "^2.2.4"
   },
   "devDependencies": {
     "@eslint/eslintrc": "^3.2.0",

pnpm-lock.yaml

@@ -10,12 +10,13 @@ import { M2MService } from './m2m.service';
 import { ChallengeApiService } from './challenge.service';
 import { EventBusService } from './eventBus.service';
 import { MemberService } from './member.service';
+import { KafkaModule } from '../kafka/kafka.module';
 
 // Global module for providing global providers
 // Add any provider you want to be global here
 @Global()
 @Module({
-  imports: [HttpModule],
+  imports: [HttpModule, KafkaModule.forRoot()],
   providers: [
     {
       provide: APP_GUARD,

src/shared/modules/global/globalProviders.module.ts

@@ -0,0 +1,24 @@
+import { LoggerService } from '../global/logger.service';
+
+export abstract class BaseEventHandler {
+  protected logger: LoggerService;
+
+  constructor(logger: LoggerService) {
+    this.logger = logger;
+  }
+
+  abstract handle(message: any): Promise<void>;
+  abstract getTopic(): string;
+
+  protected logMessage(message: any): void {
+    this.logger.log({
+      message: 'Processing Kafka message',
+      topic: this.getTopic(),
+      payload: message,
+    });
+  }
+
+  protected validateMessage(message: any): boolean {
+    return message !== null && message !== undefined;
+  }
+}

src/shared/modules/kafka/base-event.handler.ts

@@ -0,0 +1,52 @@
+import { Injectable, OnModuleInit } from '@nestjs/common';
+import { BaseEventHandler } from '../base-event.handler';
+import { KafkaHandlerRegistry } from '../kafka-handler.registry';
+import { LoggerService } from '../../global/logger.service';
+
+@Injectable()
+export class AVScanActionScanHandler
+  extends BaseEventHandler
+  implements OnModuleInit
+{
+  private readonly topic = 'avscan.action.scan';
+
+  constructor(private readonly handlerRegistry: KafkaHandlerRegistry) {
+    super(LoggerService.forRoot('AVScanActionScanHandler'));
+  }
+
+  onModuleInit() {
+    this.handlerRegistry.registerHandler(this.topic, this);
+    this.logger.log(`Registered handler for topic: ${this.topic}`);
+  }
+
+  getTopic(): string {
+    return this.topic;
+  }
+
+  async handle(message: any): Promise<void> {
+    try {
+      this.logger.log({
+        message: 'Processing AVScan Action Scan event',
+        topic: this.topic,
+        payload: message,
+      });
+
+      if (!this.validateMessage(message)) {
+        this.logger.warn('Invalid message received');
+        return;
+      }
+
+      this.logger.log('=== AVScan Action Scan Event ===');
+      this.logger.log('Topic:', this.topic);
+      this.logger.log('Payload:', JSON.stringify(message, null, 2));
+      this.logger.log('==============================');
+
+      await Promise.resolve(); // Add await to satisfy linter
+
+      this.logger.log('AVScan Action Scan event processed successfully');
+    } catch (error) {
+      this.logger.error('Error processing AVScan Action Scan event', error);
+      throw error;
+    }
+  }
+}