// Copyright (c) 2015 Uber Technologies, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.

'use strict';
/* global process */

var url = require('url');
var http = require('http');
var lbPool = require('lb_pool');
var os = require('os');
var utils = require('./utils');
/* eslint-disable no-unused-vars */
/* jshint -W098 */
var pkginfo = require('pkginfo')(module);
/* jshint +W089 */
/* eslint-enable no-unused-vars */

var supportContentType = {
    'binary': 'application/vnd.kafka.binary.v1',
    'json': 'application/vnd.kafka.json.v1+json',
    'batch': 'application/vnd.kafka.binary.batch.v1'
};

var emptyFunction = function EmptyFunction() {
};

var defaultAppNameEnvVar = 'UDEPLOY_SERVICE_NAME';
var defaultServiceNameHeader = 'kafka-rest-client-service-name';
var defaultVersionHeader = 'kafka-rest-client-version';
var defaultInstanceNameHeader = 'kafka-rest-client-instance-name';
var defaultInstanceNameEnvVar = 'UDEPLOY_DEPLOYMENT_NAME';

var defaultLocalAgentHost = 'localhost';
var defaultLocalAgentPort = 5390;
var defaultClusterUrl = 'LOGGING TOPICS';
var defaultDataClusterUrl = 'DATA TOPICS';
var hpTopicFormat = ['hp-', 'hp_', 'hp.'];
var totalWaitingTimeMs = 0;
// try connectWaitTimeMs.length times to connect to rest proxy
var connectWaitTimeMs = [0, 100, 3000, 10000, 30000, 60000, 300000, 600000];
var defaultReconnectRetryMs = 100;

String.isNullOrEmpty = function checkNullOrEmpty(value) {
    // based on https://codereview.stackexchange.com/questions/5572/string-isnullorempty-in-javascript
    return !value;
};
// Kafka 8 Rest Proxy Client.
// Initialization: takes host and port and send a GET call to get all the
// <topic, host:port> mapping.
// Produce: for sending request, forkafka client will send a POST call
function KafkaRestClient(options, callback) {
    var self = this;
    callback = callback || emptyFunction;
    self.clientType = options.clientType;
    self.enable = false;
    self.connecting = false;
    self.proxyHost = options.proxyHost;
    self.localAgentMode = options.localAgentMode;
    self.timeout = options.timeout || 1000;
    self.maxPending = options.maxPending || 500;
    self.urlToHttpClientMapping = {};
    self.cachedTopicToUrlMapping = {};
    self.proxyPort = options.proxyPort;
    self.initLineageOption(options);
    self.requestUrl = self.proxyHost + ':' + self.proxyPort;
    self.metadataWhitelist = null;
    if ('metadataWhitelist' in options) {
        self.metadataWhitelist = options.metadataWhitelist;
    }

    self.initHttpClient(options);
    self.topicDiscoveryTimes = 0;
    self.maxRetries = options.maxRetries || 3;

    self.discoverTopics(function discoverTopicsCallback(err, isDone) {
        if (err || !isDone) {
            self.enable = false;
            totalWaitingTimeMs = 0;
            self.scheduleConnect(1);
            // Will not return an error since there are retries to rest proxy
            return callback();
        }
        self.enable = true;
        return callback();
    });
    self.scheduleTopicMapRefresh();
    self.statsd = (options.statsd && typeof options.statsd.increment === 'function') ? options.statsd : null;
    self.metricsPrefix = 'kafka_rest_proxy.rest-client.' + os.hostname() + '.';
    self.localAgentHost = options.localAgentHost || defaultLocalAgentHost;
    self.localAgentPort = options.localAgentPort || defaultLocalAgentPort;
    self.produceInterval = options.produceInterval || -1;
}

KafkaRestClient.prototype.initHttpClient = function initHttpClient(options) {
    var self = this;
    if (self.clientType === 'AtLeastOnce') {
        self.mainHttpClient = new lbPool.Pool(http, [self.requestUrl], {
            /* eslint-disable camelcase */
            /* jshint camelcase: false */
            keep_alive: true,
            timeout: self.timeout,
            max_pending: self.maxPending
            /* jshint camelcase: true */
            /* eslint-enable camelcase */
        });
    } else {
        self.mainHttpClient = new lbPool.Pool(http, [self.requestUrl], {
            /* eslint-disable camelcase */
            /* jshint camelcase: false */
            keep_alive: true
            /* jshint camelcase: true */
            /* eslint-enable camelcase */
        });
    }
    self.proxyRefreshTime = options.refreshTime;

    if ('blacklistMigratorHttpClient' in options) {
        self.blacklistMigratorHttpClient = options.blacklistMigratorHttpClient;
    } else {
        self.blacklistMigratorHttpClient = null;
    }
};

KafkaRestClient.prototype.initLineageOption = function initLineageOption(options) {
    var self = this;
    if (options.serviceName) {
        // the application name for data lineage
        self.serviceName = options.serviceName;
    } else {
        if ('serviceNameEnv' in options) {
            self.serviceNameEnv = options.serviceNameEnv;
        } else {
            self.serviceNameEnv = defaultAppNameEnvVar;
        }
        /* eslint-disable no-process-env */
        self.serviceName = process.env[self.serviceNameEnv];
        /* eslint-enable no-process-env */

        if (String.isNullOrEmpty(self.serviceName)) {
            self.serviceName = utils.getServiceName();
        }
    }

    if ('serviceNameHeader' in options) {
        self.serviceNameHeader = options.serviceNameHeader;
    } else {
        self.serviceNameHeader = defaultServiceNameHeader;
    }

    if ('clientVersionHeader' in options) {
        self.clientVersionHeader = options.clientVersionHeader;
    } else {
        self.clientVersionHeader = defaultVersionHeader;
    }

    if ('instanceNameHeader' in options) {
        self.instanceNameHeader = options.instanceNameHeader;
    } else {
        self.instanceNameHeader = defaultInstanceNameHeader;
    }

    if ('instanceName' in options) {
        self.instanceName = options.instanceName;
    } else {
        if ('instanceNameEnv' in options) {
            self.instanceNameEnv = options.instanceNameEnv;
        } else {
            self.instanceNameEnv = defaultInstanceNameEnvVar;
        }
        /* eslint-disable no-process-env */
        self.instanceName = process.env[this.instanceNameEnv];
        if (String.isNullOrEmpty(self.instanceName)) {
            self.instanceName = utils.getLimitedLength(os.hostname());
        }
        /* eslint-enable no-process-env */
    }
    self.clientVersion = '[node,' + module.version + ']';

};
// produceInterval is for test purpose
// In production env, the interval is exponential back-off as 600 * Math.pow(5, retry + 1)
KafkaRestClient.prototype.getProduceInterval = function getProduceInterval(retry) {
    var self = this;
    if (self.produceInterval === -1) {
        if (self.clientType === 'AtLeastOnce') {
            return 1000;
        }
        return 600 * Math.pow(5, retry + 1);
    }
    return self.produceInterval;
};

KafkaRestClient.prototype.scheduleTopicMapRefresh = function scheduleTopicMapRefresh() {
    var self = this;
    if (self.proxyRefreshTime > 0) {
        /* eslint-disable no-undef,block-scoped-var */
        self.refreshIntervalId = setInterval(function refreshTopics() {
            totalWaitingTimeMs = 0;
            self.scheduleConnect(0);
        }, self.proxyRefreshTime);
        /* eslint-enable no-undef,block-scoped-var */
    }
};

// Reconnect to rest proxy if scheduleTopicMapRefresh fails
KafkaRestClient.prototype.scheduleConnect = function scheduleConnect(retry) {
    var self = this;
    if (retry < connectWaitTimeMs.length) {
        if (self.proxyRefreshTime > 0 && totalWaitingTimeMs + connectWaitTimeMs[retry] >= self.proxyRefreshTime) {
            return;
        }

        /* eslint-disable no-undef,block-scoped-var */
        self.reconnectId = setTimeout(function reconnect() {
            totalWaitingTimeMs += connectWaitTimeMs[retry];
            self.discoverTopics(function refreshTopicsCallback(err, isDone) {
                if (err === null && isDone) {
                    self.enable = true;
                } else if (retry < connectWaitTimeMs.length - 1) {
                    self.enable = false;
                    self.scheduleConnect(retry + 1);
                }
            });
        }, connectWaitTimeMs[retry]);
        /* eslint-enable no-undef,block-scoped-var */
    }
};

KafkaRestClient.prototype.discoverTopics = function discoverTopics(callback) {
    var self = this;
    self.connecting = true;
    this.getTopicRequestBody(self.proxyHost, self.proxyPort, function updateTopicToUrlMapping(err, res) {
        if (err) {
            self.connecting = false;
            return callback(err, false);
        }
        self.connecting = false;
        self.enable = true;
        self.cachedTopicToUrlMapping = self.getTopicToUrlMapping(res);
        self.topicDiscoveryTimes++;
        return callback(null, true);
    });
};

KafkaRestClient.prototype.getTopicRequestBody = function getTopicRequestBody(proxyHost, proxyPort, callback) {
    var self = this;
    var endpoint = '/topics';
    if (self.metadataWhitelist !== null) {
        endpoint += '?topics=' + self.metadataWhitelist.join();
    }
    self.mainHttpClient.get(endpoint, function handleGetCall(err, res, body) {
        if (err) {
            return callback(err, null);
        }
        callback(null, body);
    });
};

KafkaRestClient.prototype.getTopicToUrlMapping = function getTopicToUrlMapping(body) {
    var self = this;
    /* eslint-disable no-try-catch */
    try {
        var urlToTopicsJson = JSON.parse(body);
        var topicToUrlMapping = {};
        var urlArray = Object.keys(urlToTopicsJson);
        for (var i = 0; i < urlArray.length; i++) {
            var reqUrl = urlArray[i];
            var topicsArray = urlToTopicsJson[reqUrl];
            for (var j = 0; j < topicsArray.length; j++) {
                topicToUrlMapping[topicsArray[j]] = reqUrl;
            }
        }
        return topicToUrlMapping;
    } catch (e) {
        return self.cachedTopicToUrlMapping;
    }
    /* eslint-enable no-try-catch */
};

KafkaRestClient.prototype.isHeatPipeTopic = function isHeatPipeTopic(topic) {
    return hpTopicFormat.indexOf(topic.substring(0, 3)) > -1;
};

KafkaRestClient.prototype.goToLocalAgent = function goToLocalAgent(topic) {
    var self = this;
    return (self.clientType !== 'AtLeastOnce' && self.isHeatPipeTopic(topic)) ||
        (self.clientType === 'AtLeastOnce' && self.localAgentMode.toUpperCase() === 'ALWAYS');
};

// ProduceMessage {
// string topic,
// binary message,
// string type,
// long timeStamp
// }
KafkaRestClient.prototype.produce = function produce(produceMessage, callback) {
    var self = this;
    self.produceWithRetry(produceMessage, 0, callback);
    if (self.statsd) {
        self.statsd.increment(self.metricsPrefix + produceMessage.topic + '.produced');
    }
};

KafkaRestClient.prototype.produceWithRetry = function produceWithRetry(produceMessage, retry, callback) {
    var self = this;

    if (self.enable) {
        // push topic to whitelist if needed
        if (self.metadataWhitelist !== null && self.metadataWhitelist.indexOf(produceMessage.topic) < 0) {
            self.metadataWhitelist.push(produceMessage.topic);
        }
        if (produceMessage.topic in self.cachedTopicToUrlMapping) {
            self.tryMakeRequest(produceMessage, retry, callback);
        } else if (defaultDataClusterUrl in self.cachedTopicToUrlMapping && self.clientType !== 'AtLeastOnce') {
            self.cachedTopicToUrlMapping[produceMessage.topic] = self.cachedTopicToUrlMapping[defaultDataClusterUrl];
            self.produceWithRetry(produceMessage, 0, callback);
        } else if (self.goToLocalAgent(produceMessage.topic)) {
            self.tryMakeRequest(produceMessage, self.maxRetries - 1, callback);
        } else if (defaultClusterUrl in self.cachedTopicToUrlMapping && self.clientType !== 'AtLeastOnce') {
            self.cachedTopicToUrlMapping[produceMessage.topic] = self.cachedTopicToUrlMapping[defaultClusterUrl];
            self.produceWithRetry(produceMessage, 0, callback);
        } else if (self.clientType !== 'AtLeastOnce') {
            return callback(new Error('No default url for logging topics - ' + produceMessage.topic));
        } else {
            return callback(new Error('No url for topics - ' + produceMessage.topic));
        }
    } else if (self.connecting) {
        /* eslint-disable no-undef,block-scoped-var */
        setTimeout(function waitForInitDone() {
            self.produceWithRetry(produceMessage, retry, callback);
        }, defaultReconnectRetryMs);
        /* eslint-enable no-undef,block-scoped-var */
    } else if (self.goToLocalAgent(produceMessage.topic)) {
        self.tryMakeRequest(produceMessage, self.maxRetries - 1, callback);
    } else {
        return callback(new Error('Kafka Rest Client is not enabled yet.'));
    }
};

KafkaRestClient.prototype.tryMakeRequest = function tryMakeRequest(produceMessage, retry, callback) {
    var self = this;

    var pathUrl = self.cachedTopicToUrlMapping[produceMessage.topic];
    if (retry >= self.maxRetries - 1 && self.maxRetries > 1 && self.goToLocalAgent(produceMessage.topic)) {
        pathUrl = self.localAgentHost + ':' + self.localAgentPort;
    }
    if (pathUrl === undefined) {
        pathUrl = self.localAgentHost + ':' + self.localAgentPort;
    }

    var httpClient;
    if (self.urlToHttpClientMapping[pathUrl] === undefined) {
        if (self.clientType === 'AtLeastOnce') {
            self.urlToHttpClientMapping[pathUrl] = new lbPool.Pool(http, [pathUrl], {
                /* eslint-disable camelcase */
                /* jshint camelcase: false */
                keep_alive: true,
                timeout: self.timeout,
                max_pending: self.maxPending
                /* jshint camelcase: true */
                /* eslint-enable camelcase */
            });
        } else {
            self.urlToHttpClientMapping[pathUrl] = new lbPool.Pool(http, [pathUrl], {
                /* eslint-disable camelcase */
                /* jshint camelcase: false */
                keep_alive: true
                /* jshint camelcase: true */
                /* eslint-enable camelcase */
            });
        }
    }
    httpClient = self.urlToHttpClientMapping[pathUrl];

    var reqOpts = url.parse('http://' + pathUrl);
    reqOpts.method = 'POST';
    reqOpts.path = '/topics/' + produceMessage.topic;
    reqOpts.headers = {
        'Content-Type': supportContentType[produceMessage.type],
        'TimeStamp': produceMessage.timeStamp.toString()
    };
    if (produceMessage.key !== undefined && produceMessage.key !== '') {
        reqOpts.headers['Record-Key'] = produceMessage.key;
    }
    if (self.clientType === 'AtLeastOnce') {
        reqOpts.headers['Request-Type'] = 'sync';
    }
    reqOpts.headers[self.serviceNameHeader] = self.serviceName;
    reqOpts.headers[self.clientVersionHeader] = self.clientVersion;
    reqOpts.headers[self.instanceNameHeader] = self.instanceName;

    var time = new Date();
    httpClient.post(reqOpts, produceMessage.message, function handlePostCall(err, res, body) {
        var metricPrefix = self.metricsPrefix + produceMessage.topic;
        if (self.statsd && typeof self.statsd.timing === 'function') {
            self.statsd.timing(metricPrefix + '.latency', time);
        }

        if (err || (res && res.statusCode >= 500 && res.statusCode < 600)) {
            return self.handleError5xx(produceMessage, retry, metricPrefix, err, callback);
        }
        if (res.statusCode >= 400 && res.statusCode < 500) {
            return self.handleError4xx(produceMessage, res, callback);
        }
        if (self.clientType !== 'AtLeastOnce' ||
            (self.clientType === 'AtLeastOnce' && pathUrl !== self.localAgentHost + ':' + self.localAgentPort)) {
            return callback(null, body);
        }
        callback(new Error('Send to local agent for topic - ' + produceMessage.topic));
    });
};

/* jshint maxparams: 5 */
KafkaRestClient.prototype.handleError5xx = function handleError1(produceMessage, retry, metricPrefix, err, callback) {
    var self = this;

    if (retry < self.maxRetries - 1) {
        /* eslint-disable no-undef,block-scoped-var */
        setTimeout(function delayedProduce() {
            self.produceWithRetry(produceMessage, retry + 1, callback);
        }, self.getProduceInterval(retry));
        /* eslint-enable no-undef,block-scoped-var */

        if (self.statsd) {
            self.statsd.increment(metricPrefix + '.retry');
        }
    } else if (self.goToLocalAgent(produceMessage.topic) &&
        retry === self.maxRetries - 1 && self.maxRetries > 1) {
        /* eslint-disable no-undef,block-scoped-var */
        setTimeout(function delayedProduce() {
            self.tryMakeRequest(produceMessage, self.maxRetries, callback);
        }, 100);
        /* eslint-enable no-undef,block-scoped-var */
    } else {
        if (err !== null && typeof err === 'object') {
            // attempt field in err return by lb_pool lib has circular structure
            // will cause problem when we try to JSON.stringify it
            err.attempt = null;
        }
        if (self.statsd) {
            self.statsd.increment(metricPrefix + '.error');
        }
        return void callback(err);
    }
};

KafkaRestClient.prototype.handleError4xx = function handleError2(produceMessage, res, callback) {
    var self = this;

    if (self.goToLocalAgent(produceMessage.topic) && self.maxRetries > 1) {
        self.tryMakeRequest(produceMessage, self.maxRetries - 1, callback);
    } else {
        return void callback(new Error('Got statusCode ' + res.statusCode));
    }
};

KafkaRestClient.prototype.close = function close() {
    // Trying to close kafka rest client, need to cancel timer scheduler.
    var self = this;
    self.enable = false;
    self.connecting = false;
    if (self.proxyRefreshTime > 0) {
        // Trying to cancel TimeInterval
        /* eslint-disable no-undef,block-scoped-var */
        clearInterval(self.refreshIntervalId);
        /* eslint-enable no-undef,block-scoped-var */
        self.proxyRefreshTime = 0;
    }
    if (self.reconnectId) {
        /* eslint-disable no-undef,block-scoped-var */
        clearTimeout(self.reconnectId);
        /* eslint-enable no-undef,block-scoped-var */
    }
    self.mainHttpClient.close();
    if (self.blacklistMigratorHttpClient) {
        self.blacklistMigratorHttpClient.close();
    }
    Object.keys(self.urlToHttpClientMapping).forEach(function closeHttpClient(reqUrl) {
        self.urlToHttpClientMapping[reqUrl].close();
    });
};

module.exports = KafkaRestClient;
