'use strict';

var util = require('util');
var events = require('events');

var connstr = require('./connstr');
var auth = require('./auth');
var constants = require('./constants');

var Bucket = require('./bucket');
var ClusterManager = require('./clustermgr');
var N1qlQuery = require('./n1qlquery');
var SearchQuery = require('./searchquery');
var AnalyticsQuery = require('./analyticsquery');

function _arrayRemove(arr, element) {
  var newArray = [];
  for (var i = 0; i < arr.length; ++i) {
    if (arr[i] !== element) {
      newArray.push(arr[i]);
    }
  }
  return newArray;
}

/**
 * Represents a singular cluster containing your buckets.
 *
 * @param {string} [cnstr] The connection string for your cluster.
 * @param {Object} [options]
 *  @param {string} [options.certpath]
 *  The path to the certificate to use for SSL connections
 * @constructor
 *
 * @since 2.0.0
 * @committed
 */
function Cluster(cnstr, options) {
  this.dsnObj = connstr.parse(cnstr);
  this.auther = null;
  this.connectingBuckets = [];
  this.connectedBuckets = [];
  this.waitQueue = [];

  this.cbasHosts = null;

  // Copy passed options into the connection string
  if (options instanceof Object) {
    for (var i in options) {
      if (Object.prototype.hasOwnProperty.call(options, i)) {
        this.dsnObj.options[i] = encodeURIComponent(options[i]);
      }
    }
  }
}

/**
 * An event emitter allowing you to bind to various query result set
 * events.
 *
 * @constructor
 *
 * @private
 * @memberof Bucket
 * @extends events.EventEmitter
 *
 * @since 2.2.3
 * @uncommitted
 */
function CbasQueryResponse() {}
util.inherits(CbasQueryResponse, events.EventEmitter);
Cluster.CbasQueryResponse = CbasQueryResponse;

/**
 * Invokes an operation and dispatches a callback error if one occurs.
 *
 * @param {Function} fn The operation callback to invoke.
 * @param {Array.<*>} args An array of arguments to pass to the function.
 *
 * @private
 * @ignore
 */
Cluster.prototype._invoke = function(fn, args) {
  try {
    fn.apply(this._cb, args);
  } catch (e) {
    args[args.length - 1](e, null);
  }
};

/**
 * Will either invoke the binding method specified by fn, or alternatively
 * push the operation to a queue which is flushed once a connection
 * has been established or failed.
 *
 * @param {Function} fn The binding method to invoke.
 * @param {Array} args A list of arguments for the method.
 *
 * @private
 * @ignore
 */
Cluster.prototype._maybeInvoke = function(fn, args) {
  if (this.connectedBuckets.length > 0) {
    this._invoke(fn, args);
  } else if (this.connectingBuckets.length === 0) {
    throw new Error(
      'Cannot perform operations without at least one bucket opened.'
    );
  } else {
    this.waitQueue.push([fn, args]);
  }
};

Cluster.prototype.enableCbas = function(hosts) {
  if (!Array.isArray(hosts)) {
    hosts = [hosts];
  }

  this.cbasHosts = hosts;
};

/**
 * Specifies an authenticator to use to authenticate to this cluster.
 *
 * @param {Authenticator|Object} auther
 *
 * @since 2.1.7
 * @uncommitted
 */
Cluster.prototype.authenticate = function(auther) {
  // This implements an overload allowing specification of just a username
  // and password for RBAC based authentication.
  if (
    arguments.length === 2 &&
    typeof arguments[0] === 'string' &&
    typeof arguments[1] === 'string'
  ) {
    this.auther = new auth.PasswordAuthenticator(arguments[0], arguments[1]);
    return;
  }

  this.auther = auther;
};

/**
 * Open a bucket to perform operations.  This will begin the handshake
 * process immediately and operations will complete later.  Subscribe to
 * the `connect` event to be alerted when the connection is ready, though
 * be aware operations can be successfully queued before this.
 *
 * @param {string} [name] The name of the bucket to open.
 * @param {string} [password] Password for the bucket.
 * @param {Function} [callback]
 * Callback to invoke on connection success or failure.
 * @returns {Bucket}
 *
 * @since 2.0.0
 * @committed
 */
Cluster.prototype.openBucket = function(name, password, callback) {
  if (password instanceof Function) {
    callback = arguments[1];
    password = '';
  }

  var username = name;

  if (this.auther && password) {
    throw new Error(
      'You must not specify a bucket password when ' +
        'using cluster-level authentication.'
    );
  }

  var bucketDsnObj = connstr.normalize(this.dsnObj);
  bucketDsnObj.bucket = name;
  if (this.auther instanceof Object && this.auther.buckets instanceof Object) {
    var bucketCreds = [];
    for (var i in this.auther.buckets) {
      if (Object.prototype.hasOwnProperty.call(this.auther.buckets, i)) {
        bucketCreds.push(JSON.stringify([i, this.auther.buckets[i]]));
      }
    }
    if (bucketCreds.length > 0) {
      bucketDsnObj.options.bucket_cred = bucketCreds;
    }
  } else if (this.auther instanceof Object) {
    username = this.auther.username;
    password = this.auther.password;
  }

  if (bucketDsnObj.options.keypath) {
    if (username !== '' || password !== '') {
      throw new Error(
        'You must specify the CertAuthenticator when ' +
          'using certificate authentication.'
      );
    }
  }

  if (bucketDsnObj.options.enable_errmap === undefined) {
    bucketDsnObj.options.enable_errmap = true;
  }
  if (bucketDsnObj.options.enable_tracing === undefined) {
    bucketDsnObj.options.enable_tracing = true;
  }

  // Grab the various versions.  Note that we need to trim them
  // off as some Node.js versions insert strange characters into
  // the version identifiers (mainly newlines and such).
  var couchnodeVer = require('../package.json').version.trim();
  var nodeVer = process.versions.node.trim();
  var v8Ver = process.versions.v8.trim();
  var sslVer = process.versions.openssl.trim();

  // Generate a user-readable version string
  bucketDsnObj.options.client_string =
    'couchnode/' +
    couchnodeVer +
    ' (node/' +
    nodeVer +
    '; v8/' +
    v8Ver +
    '; ssl/' +
    sslVer +
    ')';
  this.dsnObj = bucketDsnObj;

  var bucket = new Bucket({
    dsnObj: bucketDsnObj,
    username: username,
    password: password,
  });
  if (callback) {
    bucket.on('connect', callback);
    bucket.on('error', callback);
  }

  var self = this;
  this.connectingBuckets.push(bucket);
  bucket.on('connect', function() {
    self.connectingBuckets = _arrayRemove(self.connectingBuckets, bucket);
    self.connectedBuckets.push(bucket);
    for (var i = 0; i < self.waitQueue.length; ++i) {
      var itm = self.waitQueue[i];
      self._invoke(itm[0], itm[1]);
    }
    self.waitQueue = [];
  });
  bucket.on('error', function() {
    self.connectingBuckets = _arrayRemove(self.connectingBuckets, bucket);
    if (self.connectingBuckets.length > 0) {
      return;
    }

    var err = new Error(
      'You cannot perform a cluster-level query without' +
        ' at least one bucket open.'
    );
    for (var i = 0; i < self.waitQueue.length; ++i) {
      var itm = self.waitQueue[i];
      itm[1][itm[1].length - 1](err, null);
    }
    self.waitQueue = [];
  });

  return bucket;
};

Cluster.prototype._n1qlReq = function(host, q, adhoc, emitter) {
  var bucket = this.connectedBuckets[0];
  bucket._n1qlReq(host, q, adhoc, emitter);
};

Cluster.prototype._n1ql = function(query, params, callback) {
  var req = new Bucket.N1qlQueryResponse();

  var invokeCb = callback;
  if (!invokeCb) {
    invokeCb = function(err) {
      req.emit('error', err);
    };
  }

  this._maybeInvoke(this._n1qlReq.bind(this), [
    undefined,
    query.toObject(params),
    query.isAdhoc,
    req,
    invokeCb,
  ]);

  if (callback) {
    req.on('rows', function(rows, meta) {
      callback(null, rows, meta);
    });
    req.on('error', function(err) {
      callback(err, null, null);
    });
  }

  return req;
};

Cluster.prototype._ftsReq = function(q, emitter) {
  var bucket = this.connectedBuckets[0];
  bucket._ftsReq(q, emitter);
};

Cluster.prototype._fts = function(query, callback) {
  var req = new Bucket.FtsQueryResponse();

  var invokeCb = callback;
  if (!invokeCb) {
    invokeCb = function(err) {
      req.emit('error', err);
    };
  }

  this._maybeInvoke(this._ftsReq.bind(this), [query, req, invokeCb]);

  if (callback) {
    req.on('rows', function(rows, meta) {
      callback(null, rows, meta);
    });
    req.on('error', function(err) {
      callback(err, null, null);
    });
  }

  return req;
};

Cluster.prototype._cbasReq = function(host, q, emitter) {
  var bucket = this.connectedBuckets[0];
  bucket._cbasReq(host, q, emitter);
};

Cluster.prototype._cbas = function(query, params, callback) {
  var host;
  if (this.cbasHosts) {
    var qhosts = this.cbasHosts;
    host = qhosts[Math.floor(Math.random() * qhosts.length)];
    if (host.indexOf(':') === -1) {
      host = host + ':8095';
    }
  }

  var req = new Bucket.AnalyticsQueryResponse();

  var invokeCb = callback;
  if (!invokeCb) {
    invokeCb = function(err) {
      req.emit('error', err);
    };
  }

  this._maybeInvoke(this._cbasReq.bind(this), [
    host,
    query.toObject(params),
    req,
    invokeCb,
  ]);

  if (callback) {
    req.on('rows', function(rows, meta) {
      callback(null, rows, meta);
    });
    req.on('error', function(err) {
      callback(err, null, null);
    });
  }

  return req;
};

/**
 * Executes a previously prepared query object.  This could be a
 * {@link N1qlQuery}, {@link SearchQuery} or a {@link AnalyticsQuery}.
 *
 * Note: You must have at least one bucket open (this is neccessary to
 * have cluster mapping information), and additionally be using the new
 * cluster-level authentication methods.
 *
 * Note: SearchQuery queries are currently an uncommitted interface and may be
 * subject to change in a future release.
 *
 * @param {N1qlQuery|SearchQuery|AnalyticsQuery} query
 * The query to execute.
 * @param {Object|Array} [params]
 * A list or map to do replacements on a N1QL or CBAS query.
 * @param {Bucket.QueryCallback} callback
 * @returns {Bucket.N1qlQueryResponse
 *    |Bucket.FtsQueryResponse
 *    |Cluster.AnalyticsQueryResponse}
 *
 * @since 2.1.7
 * @committed
 */
Cluster.prototype.query = function(query, params, callback) {
  if (params instanceof Function) {
    callback = arguments[1];
    params = undefined;
  }

  if (!(this.auther instanceof Object)) {
    var errC = new Error(
      'You cannot perform a cluster-level query without' +
        ' specifying credentials with Cluster.authenticate.'
    );
    setImmediate(function() {
      if (callback) {
        callback(errC);
      }
    });
    return;
  }

  if (
    this.connectedBuckets.length === 0 &&
    this.connectingBuckets.length === 0
  ) {
    var errB = new Error(
      'You cannot perform a cluster-level query without' +
        ' at least one bucket open.'
    );
    setImmediate(function() {
      if (callback) {
        callback(errB);
      }
    });
    return;
  }

  if (query instanceof N1qlQuery) {
    return this._n1ql(query, params, callback);
  } else if (query instanceof SearchQuery) {
    return this._fts(query, callback);
  } else if (query instanceof AnalyticsQuery) {
    return this._cbas(query, params, callback);
  } else {
    throw new TypeError('First argument needs to be a N1qlQuery.');
  }
};

Cluster.prototype._ftsHttpReq = function(
  method,
  username,
  password,
  path,
  contentType,
  body,
  callback
) {
  var bucket = this.connectedBuckets[0];
  return bucket._http(
    constants.ServiceType.Search,
    method,
    username,
    password,
    path,
    contentType,
    body,
    callback
  );
};

Cluster.prototype._ftsHttp = function(
  method,
  username,
  password,
  path,
  contentType,
  body,
  callback
) {
  if (username === undefined && this.auther instanceof Object) {
    username = this.auther.username;
  }
  if (password === undefined && this.auther instanceof Object) {
    password = this.auther.password;
  }

  if (
    this.connectedBuckets.length === 0 &&
    this.connectingBuckets.length === 0
  ) {
    var errB = new Error(
      'You cannot perform a cluster-level request without' +
        ' at least one bucket open.'
    );
    setImmediate(function() {
      if (callback) {
        callback(errB);
      }
    });
    return;
  }

  this._maybeInvoke(this._ftsHttpReq.bind(this), [
    method,
    username,
    password,
    path,
    contentType,
    body,
    callback,
  ]);
};

/**
 * Creates a manager allowing the management of a Couchbase cluster.
 *
 * @returns {ClusterManager}
 *
 * @since 2.0.0
 * @committed
 */
Cluster.prototype.manager = function(username, password) {
  if (username === undefined && this.auther instanceof Object) {
    username = this.auther.username;
  }
  if (password === undefined && this.auther instanceof Object) {
    password = this.auther.password;
  }
  return new ClusterManager(this, username, password);
};

module.exports = Cluster;
