import {
  Injectable,
  Logger,
  OnModuleInit,
  OnModuleDestroy,
} from '@nestjs/common';
import {
  Kafka,
  Consumer,
  EachMessagePayload,
  ConsumerSubscribeTopics,
} from 'kafkajs';
import { KafkaConfig } from './kafka.config';

export interface MessageHandler {
  topic: string;
  handler: (payload: EachMessagePayload) => Promise<void>;
}

/**
 * Kafka Consumer Service - subscribes to Kafka topics and processes messages.
 * Supports consumer groups for horizontal scaling (multiple instances).
 * Built-in graceful shutdown: commits offsets before disconnecting.
 */
@Injectable()
export class KafkaConsumerService implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger('KafkaConsumer');
  private kafka: Kafka;
  private consumer: Consumer;
  private handlers: Map<string, (payload: EachMessagePayload) => Promise<void>> =
    new Map();
  private isRunning = false;

  constructor(private readonly config: KafkaConfig) {
    if (!config.groupId) {
      throw new Error('Consumer groupId is required');
    }

    this.kafka = new Kafka({
      clientId: config.clientId,
      brokers: config.brokers,
      ssl: config.ssl ? true : undefined,
      sasl: config.sasl,
      retry: {
        retries: config.retries || 5,
        maxRetryTime: config.maxRetryTime || 30000,
      },
    });

    this.consumer = this.kafka.consumer({
      groupId: config.groupId,
      sessionTimeout: config.sessionTimeout || 30000,
      heartbeatInterval: config.heartbeatInterval || 3000,
    });
  }

  /**
   * Register a message handler for a specific topic.
   * Must be called before onModuleInit.
   */
  registerHandler(
    topic: string,
    handler: (payload: EachMessagePayload) => Promise<void>,
  ): void {
    this.handlers.set(topic, handler);
  }

  async onModuleInit() {
    if (this.handlers.size === 0) {
      this.logger.warn('No message handlers registered, skipping consumer start');
      return;
    }

    try {
      await this.consumer.connect();
      this.logger.log(
        `Kafka consumer [${this.config.groupId}] connected to [${this.config.brokers.join(', ')}]`,
      );

      const topics: ConsumerSubscribeTopics = {
        topics: Array.from(this.handlers.keys()),
        fromBeginning: false,
      };
      await this.consumer.subscribe(topics);

      this.isRunning = true;
      await this.consumer.run({
        eachMessage: async (payload: EachMessagePayload) => {
          const handler = this.handlers.get(payload.topic);
          if (handler) {
            try {
              await handler(payload);
            } catch (error) {
              this.logger.error(
                `Error processing message from ${payload.topic}[${payload.partition}]@${payload.message.offset}: ${error.message}`,
              );
              // Do not rethrow — let consumer continue processing
              // Dead-letter queue handling can be added here
            }
          }
        },
      });
    } catch (error) {
      this.logger.error(`Failed to start consumer: ${error.message}`);
      throw error;
    }
  }

  /**
   * Graceful shutdown: stop consuming, commit offsets, disconnect.
   */
  async onModuleDestroy() {
    if (this.isRunning) {
      this.logger.log(
        `Gracefully shutting down consumer [${this.config.groupId}]...`,
      );
      await this.consumer.stop();
      await this.consumer.disconnect();
      this.isRunning = false;
      this.logger.log(`Consumer [${this.config.groupId}] disconnected`);
    }
  }
}
