const Kafka = require("node-rdkafka");
const { promisify } = require("util");

/**
 * KafkaService class for handling Kafka message publishing.
 */
class KafkaService {
	/**
	 * Constructor for KafkaService.
	 * @param {Object} kafkaConfig - Kafka configuration object.
	 * @param {Object} logger - Logger instance.
	 */
	constructor(kafkaConfig, logger) {
		this.producer = new Kafka.Producer(kafkaConfig, logger);
		this.producer.connect();
		this.logger = logger;
		this.isProducingPaused = false; // Flag to control producing pause/resume
		this.bufferSize = 1000; // Maximum size of the message buffer
		this.messageBuffer = []; // Array to store buffered messages
	}

	/**
	 * Publish a message to Kafka.
	 * @param {string} topic - Kafka topic to publish the message to.
	 * @param {string} payload - Message payload to be published.
	 * @returns {Promise<void>} - A promise indicating the completion of the operation.
	 */
	async publishMessage(topic, payload) {
		await this.produceMessage(topic, payload);
	}

	/**
	 * Publish a message to Kafka with retry logic.
	 * @param {string} topic - Kafka topic to publish the message to.
	 * @param {string} payload - Message payload to be published.
	 * @param {number} retryCount - Number of retry attempts (default is 3).
	 * @returns {Promise<void>} - A promise indicating the completion of the operation.
	 */
	async publishMessageWithRetry(topic, payload, retryCount = 3) {
		for (let attempt = 1; attempt <= retryCount; attempt++) {
			try {
				await this.produceMessage(topic, payload);
				break; // Break the loop if successful
			} catch (error) {
				this.handleKafkaError(error, payload);
				if (attempt !== retryCount) {
					await new Promise((resolve) => setTimeout(resolve, 1000 * attempt));
				}
			}
		}
	}

	/**
	 * Handle Kafka errors and take appropriate actions.
	 * @param {Object} error - Kafka error object.
	 * @param {string} payload - Message payload associated with the error.
	 */
	handleKafkaError(error, payload) {
		if (error.code === Kafka.CODES.ERRORS.ERROR_NO_BROKERS_AVAILABLE) {
			this.logger.error(
				"No brokers available. Check your Kafka broker configuration.",
			);
			// Additional logic like notifying administrators can be implemented
		} else if (error.code === Kafka.CODES.ERRORS.ERROR_PRODUCER_QUEUE_FULL) {
			this.logger.error(
				"Producer queue is full. Applying backpressure strategy.",
			);
			this.pauseProducing();
		} else {
			this.logger.error("Error producing message to Kafka:", {
				error,
				payload,
			});
		}
	}

	/**
	 * Pause producing messages to Kafka to alleviate backpressure.
	 */
	pauseProducing() {
		this.logger.warn("Pausing producer to alleviate backpressure.");
		// Additional backpressure handling logic can be added here
		this.isProducingPaused = true;
		// Set a timeout to resume producing after a certain time (adjust as needed)
		setTimeout(() => {
			this.logger.warn("Resuming producer after backpressure period.");
			this.isProducingPaused = false;
		}, 5000); // Resume after 5 seconds (adjust as needed)
	}

	/**
	 * Produce a message to Kafka.
	 * @param {string} topic - Kafka topic to publish the message to.
	 * @param {string} payload - Message payload to be published.
	 * @returns {Promise<void>} - A promise indicating the completion of the operation.
	 */
	async produceMessage(topic, payload) {
		// Check if producing is paused
		if (this.isProducingPaused) {
			this.logger.warn(
				"Producing is paused. Message will be buffered or skipped.",
			);
			this.bufferMessage({ topic, payload });
			return;
		}

		const produceAsync = promisify(this.producer.produce.bind(this.producer));
		try {
			await produceAsync(topic, null, Buffer.from(payload));
			this.logger.info("Message produced to Kafka successfully.", { payload });
		} catch (error) {
			this.handleKafkaError(error, payload);
		}
	}
	/**
	 * Buffer a message when producing is paused.
	 * @param {Object} message - The message object containing topic and payload.
	 */
	bufferMessage(message) {
		// Implement  logic to buffer messages for later processing
		// For example, you can store them in an array or a dedicated buffer

		// Ensure the buffer size does not exceed the maximum limit
		if (this.messageBuffer.length < this.bufferSize) {
			this.messageBuffer.push(message);
		} else {
			// If the buffer is full, discard the oldest message
			const discardedMessage = this.messageBuffer.shift();
			this.logger.warn(
				`Buffer is full. Discarding oldest message: ${discardedMessage}`,
			);

			// Add the new message to the buffer
			this.messageBuffer.push(message);
		}

		this.logger.info("Message buffered:", message);
	}

	/**
	 * Log message send failure after multiple retries.
	 * @param {string} payload - Message payload that failed to send.
	 */
	logMessageSendFailure(payload) {
		this.logger.error(
			`Message could not be sent after multiple retries: ${payload}`,
		);
		// Additional actions or notifications can be added here
	}
}

module.exports = KafkaService;

Getting error
❯ node medics-event-publisher.js
/Users/sai/Documents/ubq/medics-event-publisher/node_modules/node-rdkafka/lib/client.js:54
  this._client = new SubClientType(globalConf, topicConf);
                 ^

Error: No such configuration property: "_readableState"
    at Producer.Client (/Users/sai/Documents/ubq/medics-event-publisher/node_modules/node-rdkafka/lib/client.js:54:18)
    at new Producer (/Users/sai/Documents/ubq/medics-event-publisher/node_modules/node-rdkafka/lib/producer.js:75:10)
    at new KafkaService (/Users/sai/Documents/ubq/medics-event-publisher/src/services/KafkaService.js:14:19)