const KafkaConnector = require("@ogury/ogury-kafka-connector");
import logger from "../utils/logger";
import { config, isUnitTest } from "../utils/config";
import { FeatureConfigService } from "./FeatureConfig.service";

export interface KafkaSerializer<T> {
  to_kafka: (data: T) => any;
}

export class KafkaService<T> {
  private readonly topic: string;
  private readonly hosts: string;
  private readonly kafkaClient;
  private featureConfigService: FeatureConfigService;

  constructor(hosts: string, topic: string, serializer?: KafkaSerializer<T>) {
    this.topic = topic;
    this.hosts = hosts;
    this.featureConfigService = FeatureConfigService.getInstance();

    if (isUnitTest()) {
      return;
    }

    // no mock, retry set to 0
    this.kafkaClient = new KafkaConnector(false, 0, serializer);

    const kafkaConfig: any = {
      "metadata.broker.list": hosts,
      "client.id": "ms-choices",
      aws_region: config.AWS_REGION,
    };

    const queueBufferingMaxKbytes = this.featureConfigService.getIntValue(
      null,
      "ms-choices.queue-buffering-max-kbytes",
      0
    );

    if (queueBufferingMaxKbytes > 0) {
      kafkaConfig["queue.buffering.max.kbytes"] = queueBufferingMaxKbytes;
    }

    this.kafkaClient.setLogger(logger);
    this.kafkaClient.setupProducer(kafkaConfig, () => {
      logger.info(`[KAFKA] producer to ${hosts} ready for topic ${this.topic}`);
    });
  }

  /**
   * Send a message to kafka
   * @param {object} message
   * @param key
   */
  send(message: T, key: string | null | undefined): void {
    try {
      this.kafkaClient.sendMessage(this.topic, message, key);
    } catch (err: any) {
      logger.error(
        err,
        `[KAFKA] producer failed to send message to topic ${this.topic}, hosts ${this.hosts}`
      );
    }
  }

  async health(): Promise<boolean> {
    return this.kafkaClient.checkHealthProducer();
  }
}