'use strict';
const fp = require('fastify-plugin');
const { Kafka } = require('kafkajs');
const tls = require('tls')
const net = require('net')
const { createTunnel, closeTunnel } = require('proxy-chain');

async function KafkaPlugin(fastify, opts, done) {
    const kafka = new Kafka({
        clientId: fastify.config.KAFKA_CLIENT_ID,
        brokers: [fastify.config.KAFKA_BOOTSTRAP_SERVER],
        //  - SASL - tested on local dev-mac, using ssl true only getting msgs
        ssl: true,
        sasl: {
            mechanism: 'plain',
            username: fastify.config.KAFKA_CLIENT_SASL_API_KEY,
            password: fastify.config.KAFKA_CLIENT_SASL_API_SECRET
        },
        socketFactory: ({ host, port, ssl, onConnect }) => {
            const socket = ssl ? new tls.TLSSocket() : new net.Socket()

            createTunnel(process.env.HTTP_PROXY, `${host}:${port}`)
                .then((tunnelAddress) => {
                    // const [tunnelHost, tunnelPort] = tunnelAddress.split(':')
                    const tunnelHost = 'localhost';
                    const tunnelPort = tunnelAddress.split(':').slice('-1')[0];

                    socket.setKeepAlive(true, 60000)
                    socket.connect(
                        Object.assign({ host: tunnelHost, port: tunnelPort, servername: host }, ssl),
                        onConnect
                    )
                })
                .catch(error => socket.emit('error', error));

            return socket;
        }
    })
    const kafkaTopic = fastify.config.KAFKA_CONSUMER_TOPIC_NAME;

    try {
        const consumer = kafka.consumer({ groupId: fastify.config.KAFKA_CLIENT_ID });
        await consumer.connect();
        fastify.log.trace(' Kafka Borker connected successfully!!')

        await consumer.subscribe({ topic: kafkaTopic, fromBeginning: true });
        fastify.log.trace(` Kafka Topic - ${kafkaTopic} subscribed successfully!!`)

        //fastify.decorate('kafka_consumer', consumer);

        await consumer.run({
            autoCommit: false,
            eachMessage: async ({ topic, partition, message }) => {
                console.log({
                    topic,
                    partition,
                    offset: message.offset,
                    produced_datetime: new Date(message.timestamp / 1000).toISOString(),
                    value: message.value.toString(),

                });

                // Publish the  messages into LBG Kafka Topic
                await fastify.lbg_kafka_producer(message.value.toString());
            },
        });
    } catch (error) {
        fastify.log.error(' Kafka broker connnection failed. So, we cannot consume messages from  Kafka Topic!!');
        console.log(error);
    }

    done();
};
module.exports = fp(KafkaPlugin, { name: 'KafkaConsumer' });