/*
  Сервис интеграции ПП Парус 8 с WEB API    
  Модуль ядра: обработчик kafka сообщений
*/

//----------------------
// Подключение библиотек
//----------------------

const _ = require("lodash"); //Работа с массивами и коллекциями
const { makeErrorText, getKafkaBroker, getKafkaAuth } = require("./utils"); //Вспомогательные функции
const { Kafka, logLevel } = require("kafkajs"); //Работа с Kafka

//------------
// Тело модуля
//------------

//Отправка сообщения Kafka
const publishKafka = async ({ connectionPrms, url, auth, topic, message }) => {
    //Иницализируем подключение к Kafka
    let kafka = new Kafka({
        clientId: connectionPrms.sClientIdSender,
        brokers: [url],
        connectionTimeout: connectionPrms.nConnectionTimeout,
        logLevel: logLevel.NOTHING,
        ...auth
    });
    //Инициализируем продюсера
    let producer = kafka.producer();
    //Подключаемся к Kafka
    await producer.connect();
    //Отправляем сообщение
    let res = await producer.send({ topic: topic, messages: [{ value: message }] });
    //Отключаемся
    await producer.disconnect();
    //Возвращаем ответ
    return res;
};

//Получение MQTT сообщений
const subscribeKafka = async ({ connectionPrms, service, processKafkaMessage, logger }) => {
    try {
        //Признак необходимости вывода сообщения о потере соединения
        let bLogLostConnection = true;
        //Получаем брокера по URL сервиса
        let sBroker = getKafkaBroker(service.sSrvRoot);
        //Иницализируем подключение к Kafka
        let client = new Kafka({
            clientId: connectionPrms.sClientIdRecipient,
            brokers: [sBroker],
            connectionTimeout: connectionPrms.nConnectionTimeout,
            ...getKafkaAuth(service.sSrvUser, service.sSrvPass),
            logLevel: logLevel.NOTHING,
            retry: {
                retries: 0,
                maxRetryTime: connectionPrms.nMaxRetryTime,
                initialRetryTime: connectionPrms.nInitialRetryTime,
                restartOnFailure: error => {
                    return new Promise(resolve => {
                        //Если требуется вывести ошибку
                        if (bLogLostConnection) {
                            //Выводим ошибку
                            logger.error(`Соединение с Kafka потеряно (${sBroker}): ${makeErrorText(error)}`);
                            //Сбрасываем признак необходимости вывода ошибки
                            bLogLostConnection = false;
                        }
                        resolve(connectionPrms.bRestartOnFailure);
                    });
                }
            }
        });

        //Инициализируем получателя
        let consumer = client.consumer({ groupId: "ParusWebApi" });

        //Устанавливаем прослушивание
        await consumer.connect();
        consumer.subscribe({ topics: _.map(service.functions, "sFnURL") });

        //Запускаем прослушивание необходимых топиков
        consumer.run({
            eachMessage: async ({ topic, message }) => {
                try {
                    //Вызываем обработчик
                    processKafkaMessage({
                        message,
                        service,
                        fn: _.find(service.functions, { sFnURL: topic })
                    });
                } catch (e) {
                    await logger.error(`Ошибка обработки исходящего сообщения Kafka: ${makeErrorText(e)}`);
                }
            }
        });

        //Отслеживаем соединение
        consumer.on(consumer.events.CONNECT, () => {
            //Если сообщение о потере соединения уже выводилось
            if (!bLogLostConnection) {
                //Сообщим о восстановлении соединения
                logger.info(`Соединение с Kafka восстановлено (${sBroker})`);
                //Устанавливаем признак сообщения о потере соединения
                bLogLostConnection = true;
            }
        });
        //Возвращаем соединение
        return consumer;
    } catch (e) {
        await logger.error(`Ошибка запуска обработчика очереди исходящих сообщений Kafka: ${makeErrorText(e)}`);
    }
};

//-----------------
// Интерфейс модуля
//-----------------

exports.publishKafka = publishKafka;
exports.subscribeKafka = subscribeKafka;