import logger from "../../utils/logger";
import { config } from "../../utils/config";

const kafka = require("@ogury/ogury-kafka-connector");

export class KafkaConnector {
  kafkaClient;
  readyPromise: Promise<void>;

  constructor(private readonly hosts: string, private readonly topic: string) {
    const kafkaConfig = {
      "metadata.broker.list": hosts,
      "client.id": "ms-choices",
      aws_region: config.AWS_REGION,
    };

    this.kafkaClient = new kafka();
    this.kafkaClient.setLogger(logger);
    const producer = this.kafkaClient.setupProducer(kafkaConfig);

    this.readyPromise = new Promise((resolve, reject) => {
      producer.on("error", reject);

      const readyCallback = () => {
        producer.removeListener("error", reject);

        logger.info(`[KAFKA] producer to ${hosts} ready for topic ${this.topic}`);
        resolve();
      };
      producer.on("ready", readyCallback);
    });
  }

  send(message: any, key?: string) {
    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}`
      );
    }
  }
}