"use strict";
var assert = require('assert');
var util = require('util');

var helper = require('../../test-helper');
var Client = require('../../../lib/client');
var Connection = require('../../../lib/connection');
var utils = require('../../../lib/utils');
var types = require('../../../lib/types');
var errors = require('../../../lib/errors');
var ExecutionProfile = require('../../../lib/execution-profile').ExecutionProfile;
var loadBalancing = require('../../../lib/policies').loadBalancing;
var vdescribe = helper.vdescribe;

describe('client read timeouts', function () {
  this.timeout(120000);
  before(helper.ccmHelper.start(2));
  after(helper.ccmHelper.remove);
  afterEach(function (done) {
    // Tests will pause any of the nodes and should resume it in the general case, but if for whatever reason they fail
    // resuming the nodes should be safe.
    utils.series([
      helper.toTask(helper.ccmHelper.resumeNode, null, 1),
      helper.toTask(helper.ccmHelper.resumeNode, null, 2)
    ], done);
  });
  describe('when socketOptions.readTimeout is not set', function () {
    it('should do nothing else than waiting', getTimeoutErrorNotExpectedTest(false, false));
    it('should use readTimeout when defined', getMoveNextHostTest(false, false, 3123, 0, { readTimeout: 3123 }));
  });
  describe('when socketOptions.readTimeout is set', function () {
    it('should move to next host by default for simple queries', getMoveNextHostTest(false, false));
    it('should move to next host for prepared queries executions', getMoveNextHostTest(true, true));
    it('should move to next host for prepared requests', getMoveNextHostTest(true, false));
    it('should move to next host for the initial prepare', getMoveNextHostTest(true, false));
    it('should callback in error when retryOnTimeout is false', function (done) {
      var client = newInstance({ socketOptions: { readTimeout: 3000 } });
      var coordinators = {};
      var errorsReceived = [];
      utils.series([
        client.connect.bind(client),
        function warmup(next) {
          utils.timesSeries(10, function (n, timesNext) {
            client.execute('SELECT key FROM system.local', function (err, result) {
              if (err) {
                return timesNext(err);
              }
              coordinators[helper.lastOctetOf(result.info.queriedHost)] = true;
              timesNext();
            });
          }, next);
        },
        helper.toTask(helper.ccmHelper.pauseNode, null, 2),
        function checkTimeouts(next) {
          assert.strictEqual(Object.keys(coordinators).length, 2);
          assert.strictEqual(coordinators['1'], true);
          assert.strictEqual(coordinators['2'], true);
          coordinators = {};
          utils.times(10, function (n, timesNext) {
            client.execute('SELECT key FROM system.local', [], { retryOnTimeout: false }, function (err, result) {
              if (err) {
                errorsReceived.push(err);
              }
              else {
                coordinators[helper.lastOctetOf(result.info.queriedHost)] = true;
              }
              timesNext();
            });
          }, function (err) {
            if (err) {
              return next(err);
            }
            assert.strictEqual(Object.keys(coordinators).length, 1);
            assert.strictEqual(coordinators['1'], true);
            //half of the executions failed
            assert.strictEqual(errorsReceived.length, 5);
            assert.ok(errorsReceived.reduce(function (previous, current) {
              return previous && (current instanceof errors.OperationTimedOutError);
            }, true));
            next();
          });
        },
        helper.toTask(helper.ccmHelper.resumeNode, null, 2),
        client.shutdown.bind(client)
      ], done);
    });
    it('defunct the connection when the threshold passed', function (done) {
      var client = newInstance({
        socketOptions: {
          readTimeout: 3000,
          defunctReadTimeoutThreshold: 32
        },
        //1 connection per host to simply it
        poolingOptions: {
          coreConnectionsPerHost: {
            '0': 1,
            '1': 1,
            '2': 0
          }
        }
      });
      var coordinators = {};
      var connection;
      utils.series([
        client.connect.bind(client),
        function warmup(next) {
          utils.times(10, function (n, timesNext) {
            client.execute('SELECT key FROM system.local', function (err, result) {
              if (err) {
                return timesNext(err);
              }
              coordinators[helper.lastOctetOf(result.info.queriedHost)] = true;
              timesNext();
            });
          }, next);
        },
        function identifyConnection(next) {
          connection = client.hosts.values()
            .filter(function (h) {
              return helper.lastOctetOf(h) === '2';
            })[0]
            .pool
            .connections[0];
          helper.assertInstanceOf(connection, Connection);
          assert.strictEqual(connection.netClient.writable, true);
          next();
        },
        helper.toTask(helper.ccmHelper.pauseNode, null, 2),
        function checkTimeouts(next) {
          assert.strictEqual(Object.keys(coordinators).length, 2);
          assert.strictEqual(coordinators['1'], true);
          assert.strictEqual(coordinators['2'], true);
          coordinators = {};
          utils.times(500, function (n, timesNext) {
            client.execute('SELECT key FROM system.local', function (err, result) {
              if (err) {
                return timesNext(err);
              }
              coordinators[helper.lastOctetOf(result.info.queriedHost)] = true;
              timesNext();
            });
          }, function (err) {
            if (err) {
              return next(err);
            }
            assert.strictEqual(Object.keys(coordinators).length, 1);
            assert.strictEqual(coordinators['1'], true);
            //we check using an internal property of socket which is lame
            //and might break
            assert.strictEqual(connection.netClient.writable, false);
            next();
          });
        },
        helper.toTask(helper.ccmHelper.resumeNode, null, 2),
        client.shutdown.bind(client)
      ], done);
    });
    it('should move to next host for eachRow() executions', function (done) {
      var client = newInstance({ socketOptions: { readTimeout: 3000 } });
      var coordinators = {};
      utils.series([
        client.connect.bind(client),
        function warmup(next) {
          utils.timesSeries(10, function (n, timesNext) {
            var counter = 0;
            client.eachRow('SELECT key FROM system.local', [], function () {
              counter++;
            }, function (err, result) {
              if (err) {
                return timesNext(err);
              }
              coordinators[helper.lastOctetOf(result.info.queriedHost)] = true;
              assert.strictEqual(result.rowLength, counter);
              timesNext();
            });
          }, next);
        },
        helper.toTask(helper.ccmHelper.pauseNode, null, 2),
        function checkTimeouts(next) {
          assert.strictEqual(Object.keys(coordinators).length, 2);
          assert.strictEqual(coordinators['1'], true);
          assert.strictEqual(coordinators['2'], true);
          coordinators = {};
          utils.times(10, function (n, timesNext) {
            var counter = 0;
            client.eachRow('SELECT key FROM system.local', [], function () {
              counter++;
            }, function (err, result) {
              if (err) {
                return timesNext(err);
              }
              coordinators[helper.lastOctetOf(result.info.queriedHost)] = true;
              assert.strictEqual(result.rowLength, counter);
              timesNext();
            });
          }, function (err) {
            if (err) {
              return next(err);
            }
            assert.strictEqual(Object.keys(coordinators).length, 1);
            assert.strictEqual(coordinators['1'], true);
            next();
          });
        },
        helper.toTask(helper.ccmHelper.resumeNode, null, 2),
        client.shutdown.bind(client)
      ], done);
    });
  });
  describe('when queryOptions.readTimeout is set', function () {
    function profiles() {
      return [
        new ExecutionProfile('aProfile', {readTimeout: 8675})
      ];
    }
    it('should be used instead of socketOptions.readTimeout and profile.readTimeout for simple queries',
      getMoveNextHostTest(false, false, 3123, 1 << 24, { executionProfile: 'aProfile', readTimeout: 3123 }, profiles()));
    it('should be used instead of socketOptions.readTimeout and profile.readTimeout for prepared queries executions',
      getMoveNextHostTest(true, true, 3123, 1 << 24, { executionProfile: 'aProfile', readTimeout: 3123 }, profiles()));
    it('should suppress socketOptions.readTimeout and profile.readTimeout when set to 0 for simple queries',
      getTimeoutErrorNotExpectedTest(false, false, 1000, { executionProfile: 'aProfile', readTimeout: 0}, profiles()));
    it('should suppress socketOptions.readTimeout and profile.readTimeout when set to 0 for prepared queries executions',
      getTimeoutErrorNotExpectedTest(true, true, 1000, { executionProfile: 'aProfile', readTimeout: 0}, profiles()));
  });
  describe('when executionProfile.readTimeout is set', function() {
    function timeoutProfiles() {
      return [
        new ExecutionProfile('indefiniteTimeout', {readTimeout: 0}),
        new ExecutionProfile('definedTimeout', {readTimeout: 3123})
      ];
    }
    it('should be used instead of socketOptions.readTimeout for simple queries',
      getMoveNextHostTest(false, false, 3123, 1 << 24, { executionProfile: 'definedTimeout' }, timeoutProfiles()));
    it('should be used instead of socketOptions.readTimeout for prepared queries executions',
      getMoveNextHostTest(true, true, 3123, 1 << 24, { executionProfile: 'definedTimeout' }, timeoutProfiles()));
    it('should suppress socketOptions.readTimeout when set to 0 for simple queries',
      getTimeoutErrorNotExpectedTest(false, false, 1000, { executionProfile: 'indefiniteTimeout'}, timeoutProfiles()));
    it('should suppress socketOptions.readTimeout when set to 0 for prepared queries executions',
      getTimeoutErrorNotExpectedTest(true, true, 1000, { executionProfile: 'indefiniteTimeout'}, timeoutProfiles()));
  });
  vdescribe('2.0', 'with prepared batches', function () {
    it('should retry when preparing multiple queries', function (done) {
      var client = newInstance({
        // Use a lbp that always yields the hosts in the same order
        policies: { loadBalancing: new FixedOrderLoadBalancingPolicy() },
        pooling: { warmup: true, coreConnectionsPerHost: { '0': 1 }},
        socketOptions: { readTimeout: 1000 },
        queryOptions: { consistency: types.consistencies.one }
      });
      var ksName = 'ks_batch_test1';
      var tableName = ksName + '.table1';
      utils.series([
        client.connect.bind(client),
        helper.toTask(client.execute, client, helper.createKeyspaceCql(ksName, 2)),
        helper.toTask(client.execute, client, helper.createTableCql(tableName)),
        helper.toTask(helper.ccmHelper.pauseNode, null, 1),
        function checkPreparing(next) {
          var queries = [{
            query: util.format("INSERT INTO %s (id, text_sample) VALUES (?, ?)", tableName),
            params: [types.Uuid.random(), 'one']
          }, {
            query: util.format("INSERT INTO %s (id, int_sample) VALUES (?, ?)", tableName),
            params: [types.Uuid.random(), 2]
          }];
          // It should be retried on the next node
          client.batch(queries, { prepare: true, logged: false }, function (err, result) {
            assert.ifError(err);
            assert.strictEqual(helper.lastOctetOf(result.info.queriedHost), '2');
            next();
          });
        },
        helper.toTask(helper.ccmHelper.resumeNode, null, 1),
        client.shutdown.bind(client)
      ], done);
    });
    it('should produce a NoHostAvailableError when prepare tried and timed out on all hosts', function (done) {
      var client = newInstance({
        // Use a lbp that always yields the hosts in the same order
        policies: { loadBalancing: new FixedOrderLoadBalancingPolicy() },
        pooling: { warmup: true, coreConnectionsPerHost: { '0': 1 }},
        socketOptions: { readTimeout: 1000 },
        queryOptions: { consistency: types.consistencies.one }
      });
      var ksName = 'ks_batch_test2';
      var tableName = ksName + '.table1';
      utils.series([
        client.connect.bind(client),
        helper.toTask(client.execute, client, helper.createKeyspaceCql(ksName, 2)),
        helper.toTask(client.execute, client, helper.createTableCql(tableName)),
        helper.toTask(helper.ccmHelper.pauseNode, null, 1),
        helper.toTask(helper.ccmHelper.pauseNode, null, 2),
        function checkPreparing(next) {
          var queries = [{
            query: util.format("INSERT INTO %s (id, text_sample) VALUES (?, ?)", tableName),
            params: [types.Uuid.random(), 'one']
          }, {
            query: util.format("INSERT INTO %s (id, int_sample) VALUES (?, ?)", tableName),
            params: [types.Uuid.random(), 2]
          }];
          // It should be tried on all nodes and produce a NoHostAvailableError.
          client.batch(queries, { prepare: true, logged: false }, function (err) {
            helper.assertInstanceOf(err, errors.NoHostAvailableError);
            //noinspection JSUnresolvedVariable
            var numErrors = Object.keys(err.innerErrors).length;
            assert.strictEqual(numErrors, 2);
            next();
          });
        },
        helper.toTask(helper.ccmHelper.resumeNode, null, 1),
        helper.toTask(helper.ccmHelper.resumeNode, null, 2),
        client.shutdown.bind(client)
      ], done);
    });
  });
});


/** @returns {Client}  */
function newInstance(options) {
  return new Client(utils.deepExtend({}, helper.baseOptions, options));
}

/**
 * @param {Boolean} prepare
 * @param {Boolean} prepareWarmup
 * @param {Number} [expectedTimeoutMillis]
 * @param {Number} [readTimeout]
 * @param {QueryOptions} [queryOptions]
 * @param {Array.<ExecutionProfile>} [profiles]
 * @returns {Function}
 */
function getMoveNextHostTest(prepare, prepareWarmup, expectedTimeoutMillis, readTimeout, queryOptions, profiles) {
  if (!expectedTimeoutMillis) {
    expectedTimeoutMillis = 3000;
  }
  if (!readTimeout) {
    readTimeout = 3000;
  }
  profiles = profiles || [];
  return (function moveNextHostTest(done) {
    var client = newInstance({ profiles: profiles, socketOptions: { readTimeout: readTimeout } });
    var timeoutLogs = [];
    client.on('log', function (level, constructorName, info) {
      if (level !== 'warning' || info.indexOf('timeout') === -1) {
        return;
      }
      timeoutLogs.push(info);
    });
    var coordinators = {};
    utils.series([
      client.connect.bind(client),
      function warmup(next) {
        utils.timesSeries(10, function (n, timesNext) {
          client.execute('SELECT key FROM system.local', [], { prepare: prepareWarmup }, function (err, result) {
            if (err) {
              return timesNext(err);
            }
            coordinators[helper.lastOctetOf(result.info.queriedHost)] = true;
            timesNext();
          });
        }, next);
      },
      helper.toTask(helper.ccmHelper.pauseNode, null, 2),
      function checkTimeouts(next) {
        assert.strictEqual(Object.keys(coordinators).length, 2);
        assert.strictEqual(coordinators['1'], true);
        assert.strictEqual(coordinators['2'], true);
        coordinators = {};
        var testAbortTimeout = setTimeout(function () {
          throw new Error('It should have been executed in the next (not paused) host.');
        }, expectedTimeoutMillis * 4);
        utils.times(10, function (n, timesNext) {
          queryOptions = utils.extend({ }, queryOptions, { prepare: prepare});
          client.execute('SELECT key FROM system.local', [], queryOptions, function (err, result) {
            if (err) {
              return timesNext(err);
            }
            coordinators[helper.lastOctetOf(result.info.queriedHost)] = true;
            timesNext();
          });
        }, function timeFinished(err) {
          clearTimeout(testAbortTimeout);
          if (err) {
            return next(err);
          }
          assert.strictEqual(Object.keys(coordinators).length, 1);
          assert.strictEqual(coordinators['1'], true);
          assert.ok(timeoutLogs.length);
          //check that the logs messages contains the actual millis value
          assert.ok(timeoutLogs.reduce(function (val, current) {
            return val || current.indexOf(expectedTimeoutMillis.toString()) >= 0;
          }, false), 'Timeout millis not found');
          next();
        });
      },
      helper.toTask(helper.ccmHelper.resumeNode, null, 2),
      client.shutdown.bind(client)
    ], done);
  });
}

function getTimeoutErrorNotExpectedTest(prepare, prepareWarmup, readTimeout, queryOptions, profiles) {
  if (typeof readTimeout === 'undefined') {
    readTimeout = 0;
  }

  profiles = profiles || [];

  return (function timeoutErrorNotExpectedTest(done) {
    var client = newInstance({ profiles: profiles, socketOptions: { readTimeout: readTimeout } });
    var coordinators = {};
    utils.series([
      client.connect.bind(client),
      function warmup(next) {
        utils.timesSeries(10, function (n, timesNext) {
          client.execute('SELECT key FROM system.local', [], { prepare: prepareWarmup }, function (err, result) {
            if (err) {
              return timesNext(err);
            }
            coordinators[helper.lastOctetOf(result.info.queriedHost)] = true;
            timesNext();
          });
        }, next);
      },
      helper.toTask(helper.ccmHelper.pauseNode, null, 2),
      function checkTimeouts(next) {
        assert.strictEqual(Object.keys(coordinators).length, 2);
        assert.strictEqual(coordinators['1'], true);
        assert.strictEqual(coordinators['2'], true);
        coordinators = {};
        //execute 2 queries without waiting for the response
        var cb = function (err, result) {
          assert.ifError(err);
          coordinators[helper.lastOctetOf(result.info.queriedHost)] = true;
        };
        for (var i = 0; i < 2; i++) {
          queryOptions = utils.extend({ }, queryOptions, { prepare: prepare });
          client.execute('SELECT key FROM system.local', [], queryOptions, cb);
        }
        //wait for the node that is healthy to respond
        setTimeout(next, 2000);
      },
      function checkWhenPaused(next) {
        //the other callback is still waiting
        assert.strictEqual(Object.keys(coordinators).length, 1);
        assert.strictEqual(coordinators['1'], true);
        next();
      },
      helper.toTask(helper.ccmHelper.resumeNode, null, 2),
      function waitForResponse(next) {
        // Wait for 2 seconds after resume for node to respond.
        setTimeout(next, 2000);
      },
      function checkAfterResuming(next) {
        // Should get responses from each coordinator since no requests should have timed out.
        assert.strictEqual(Object.keys(coordinators).length, 2);
        assert.strictEqual(coordinators['1'], true);
        assert.strictEqual(coordinators['2'], true);
        next();
      },
      client.shutdown.bind(client)
    ], done);
  });
}

/**
 * Represents a LoadBalancingPolicy that always yields the hosts in the same order, only suitable for testing.
 * @constructor
 */
function FixedOrderLoadBalancingPolicy() {
}

util.inherits(FixedOrderLoadBalancingPolicy, loadBalancing.RoundRobinPolicy);

FixedOrderLoadBalancingPolicy.prototype.newQueryPlan = function (ks, q, callback) {
  callback(null, utils.arrayIterator(this.hosts.values()));
};