/**
 * Module dependencies
 */
var Connection = require('./connection')
  , events = require('events')
  , util = require('util')
  , crypto = require('crypto');

/**
 * Default options
 */
var defaultConnectOptions = {
  keepalive: 60,
  protocolId: 'MQIsdp',
  protocolVersion: 3,
  reconnectPeriod: 1000,
  clean: true,
  encoding: 'utf8'
};

var defaultId = function() {
  return 'mqttjs_' + crypto.randomBytes(8).toString('hex');
};

var nop = function(){};

/**
 * MqttClient constructor
 *
 * @param {Stream} stream - stream
 * @param {Object} [options] - connection options
 * (see Connection#connect)
 */
var MqttClient = module.exports =
function MqttClient(streamBuilder, options) {
  var that = this;

  if (!(this instanceof MqttClient)) {
    return new MqttClient(streamBuilder, options);
  }

  this.options = options || {};

  // Defaults
  for(var k in defaultConnectOptions) {
    if ('undefined' === typeof this.options[k]) {
      this.options[k] = defaultConnectOptions[k];
    } else {
      this.options[k] = options[k];
    }
  }

  this.options.clientId = this.options.clientId || defaultId();

  this.streamBuilder = streamBuilder;

  this._setupStream();

  // Ping timer, setup in _setupPingTimer
  this.pingTimer = null;
  // Is the client connected?
  this.connected = false;
  // Packet queue
  this.queue = [];
  // Are we intentionally disconnecting?
  this.disconnecting = false;
  // Reconnect timer
  this.reconnectTimer = null;
  // MessageIDs starting with 1
  this.nextId = Math.floor(Math.random() * 65535);

  // Inflight messages
  this.inflight = {
    puback: {},
    pubrec: {},
    pubcomp: {},
    suback: {},
    unsuback: {}
  };

  // Incoming messages
  this.incoming = {
    pubrel: {}
  };

  // Mark connected on connect
  this.on('connect', function() {
    this.connected = true;
  });

  // Mark disconnected on stream close
  this.on('close', function() {
    this.connected = false;
  });

  // Setup ping timer
  this.on('connect', this._setupPingTimer);

  // Send queued packets
  this.on('connect', function() {
    var queue = that.queue
      , length = queue.length;

    for (var i = 0; i < length; i += 1) {
      that._sendPacket(
        queue[i].type,
        queue[i].packet,
        queue[i].cb
      );
    }
    that.queue = [];
  });


  // Clear ping timer
  this.on('close', function () {
    if (that.pingTimer !== null) {
      clearInterval(that.pingTimer);
      that.pingTimer = null;
    }
  });

  // Setup reconnect timer on disconnect
  this.on('close', function() {
    that._setupReconnect();
  });

  events.EventEmitter.call(this);
};
util.inherits(MqttClient, events.EventEmitter);

/**
 * setup the event handlers in the inner stream.
 *
 * @api private
 */
MqttClient.prototype._setupStream = function() {
  var that = this;

  this._clearReconnect();

  this.stream = this.streamBuilder();

  // MqttConnection
  this.conn = this.stream.pipe(new Connection());

  // Set encoding of incoming publish payloads
  if (this.options.encoding) {
    this.conn.setPacketEncoding(this.options.encoding);
  }

  // Suppress connection errors
  this.stream.on('error', nop)

  // Echo stream close
  this.stream.on('close', this.emit.bind(this, 'close'));

  // Send a connect packet on stream connect
  this.stream.on('connect', function () {
    that.conn.connect(that.options);
  });

  this.stream.on('secureConnect', function () {
    that.conn.connect(that.options);
  });

  // Handle incoming publish
  this.conn.on('publish', function (packet) {
    that._handlePublish(packet);
  });

  // one single handleAck function
  var handleAck = function (packet) {
    that._handleAck(packet);
  };

  // Handle incoming acks
  var acks = ['puback', 'pubrec', 'pubcomp', 'suback', 'unsuback'];

  acks.forEach(function (event) {
    that.conn.on(event, handleAck);
  });

  // Handle outgoing acks
  this.conn.on('pubrel', function (packet) {
    that._handlePubrel(packet);
  });

  // Handle connack
  this.conn.on('connack', function (packet) {
    that._handleConnack(packet);
  });

  // Handle pingresp
  this.conn.on('pingresp', function (packet) {
    that._handlePingresp(packet);
  });

  // Echo connection errors
  this.conn.on('error', this.emit.bind(this, 'error'));
};

/**
 * publish - publish <message> to <topic>
 *
 * @param {String} topic - topic to publish to
 * @param {String, Buffer} message - message to publish
 * @param {Object} [opts] - publish options, includes:
 *    {Number} qos - qos level to publish on
 *    {Boolean} retain - whether or not to retain the message
 * @param {Function} [callback] - function(err){}
 *    called when publish succeeds or fails
 * @returns {MqttClient} this - for chaining
 * @api public
 *
 * @example client.publish('topic', 'message');
 * @example
 *     client.publish('topic', 'message', {qos: 1, retain: true});
 * @example client.publish('topic', 'message', console.log);
 */
MqttClient.prototype.publish =
function(topic, message, opts, callback) {
  var packet;

  // .publish(topic, payload, cb);
  if ('function' === typeof opts) {
    callback = opts;
    opts = null;
  }

  // Default opts
  if(!opts) opts = {qos: 0, retain: false};

  callback = callback || nop;

  packet = {
    topic: topic,
    payload: message,
    qos: opts.qos,
    retain: opts.retain,
    messageId: this._nextId()
  };

  this._sendPacket('publish', packet, function () {
    switch (opts.qos) {
      case 0:
        // Immediately callback
        callback();
        break;
      case 1:
        // Add to puback callbacks
        this.inflight.puback[packet.messageId] = callback;
        break;
      case 2:
        // Add to pubrec callbacks
        this.inflight.pubrec[packet.messageId] = callback;
        break;
      default:
        break;
    }
  });

  return this;
};

/**
 * subscribe - subscribe to <topic>
 *
 * @param {String, Array, Object} topic - topic(s) to subscribe to, supports objects in the form {'topic': qos}
 * @param {Object} [opts] - optional subscription options, includes:
 *    {Number} qos - subscribe qos level
 * @param {Function} [callback] - function(err, granted){} where:
 *    {Error} err - subscription error (none at the moment!)
 *    {Array} granted - array of {topic: 't', qos: 0}
 * @returns {MqttClient} this - for chaining
 * @api public
 * @example client.subscribe('topic');
 * @example client.subscribe('topic', {qos: 1});
 * @example client.subscribe({'topic': 0, 'topic2': 1}, console.log);
 * @example client.subscribe('topic', console.log);
 */
MqttClient.prototype.subscribe =
function() {
  var args = Array.prototype.slice.call(arguments)
    , subs = [];
  
  var obj = args.shift()
    , callback = args.pop() || nop
    , opts = args.pop();
  
  obj = typeof obj == 'string'? [obj] : obj;
  if (toString.call(callback) == '[object Object]') {
    opts = callback;
    callback = nop;
  }
  
  if (!opts) {
    opts = {qos: 0};
  }
  
  if (!(toString.call(obj) == '[object Object]' || util.isArray(obj))) {
    throw 'Invalid topic, or topics list';
  }
  
  if (util.isArray(obj)) {
    obj.forEach(function(topic) {
      subs.push({
	topic: topic,
	qos: opts.qos
      });
    });
  } else {
    for (var k in obj) {
      if (obj.hasOwnProperty(k)) {
	subs.push({
	  topic: k,
	  qos: obj[k]
	});
      }
    }
  }

  var packet = {
    subscriptions: subs,
    qos: 1,
    retain: false,
    dup: false,
    messageId: this._nextId()
  };

  this._sendPacket('subscribe', packet, function () {
    this.inflight.suback[packet.messageId] = {
      callback: callback,
      packet: packet
    };
  });

  return this;
};

/**
 * unsubscribe - unsubscribe from topic(s)
 *
 * @param {String, Array} topic - topics to unsubscribe from
 * @param {Function} [callback] - callback fired on unsuback
 * @returns {MqttClient} this - for chaining
 * @api public
 * @example client.unsubscribe('topic');
 * @example client.unsubscribe('topic', console.log);
 */
MqttClient.prototype.unsubscribe = function(topic, callback) {
  callback = callback || nop;
  var packet = {messageId: this._nextId()};

  if ('string' === typeof topic) {
    packet.unsubscriptions = [topic];
  } else if ('object' === typeof topic && topic.length) {
    packet.unsubscriptions = topic;
  }

  this._sendPacket('unsubscribe', packet, function () {
    this.inflight.unsuback[packet.messageId] = callback;
  });

  return this;
};

/**
 * end - close connection
 *
 * @returns {MqttClient} this - for chaining
 * @api public
 */
MqttClient.prototype.end = function() {
  this.disconnecting = true;
  if (!this.connected) {
    this.once('connect', this._cleanUp.bind(this));
  } else {
    this._cleanUp();
  }

  return this;
};

/**
 * _reconnect - implement reconnection
 * @api privateish
 */
MqttClient.prototype._reconnect = function() {

  this.emit('reconnect');

  if (this.conn) {
    this.conn.removeAllListeners();
    delete this.conn;
  }

  this._setupStream();
};

/**
 * _setupReconnect - setup reconnect timer
 */
MqttClient.prototype._setupReconnect = function() {
  var that = this;

  if (!that.disconnecting && !that.reconnectTimer && (that.options.reconnectPeriod > 0)) {
    this.emit('offline');
    that.reconnectTimer = setInterval(function () {
      that._reconnect();
    }, that.options.reconnectPeriod);
  }
};

/**
 * _clearReconnect - clear the reconnect timer
 */
MqttClient.prototype._clearReconnect = function() {
  if (this.reconnectTimer) {
    clearInterval(this.reconnectTimer);
    this.reconnectTimer = false;
  }
};


/**
 * _cleanUp - clean up on connection end
 * @api private
 */
MqttClient.prototype._cleanUp = function(forced) {
  this.conn.disconnect();
  if (forced) {
    this.stream.destroy();
  } else {
    this.stream.end();
  }
  if (this.pingTimer !== null) {
    clearInterval(this.pingTimer);
    this.pingTimer = null;
  }
};

/**
 * _sendPacket - send or queue a packet
 * @param {String} type - packet type (see `protocol`)
 * @param {Object} packet - packet options
 * @param {Function} cb - callback when the packet is sent
 * @api private
 */

MqttClient.prototype._sendPacket = function(type, packet, cb) {
  if (this.connected) {
    this.conn[type](packet);
    if (cb) cb.call(this);
  } else {
    this.queue.push({type: type, packet: packet, cb: cb});
  }
};

/**
 * _setupPingTimer - setup the ping timer
 *
 * @api private
 */
MqttClient.prototype._setupPingTimer = function() {
  var that = this;

  if (!this.pingTimer && this.options.keepalive) {
    this.pingResp = true;
    this.pingTimer = setInterval(function () {
        that._checkPing();
    }, this.options.keepalive * 1000);
  }
};

/**
 * _checkPing - check if a pingresp has come back, and ping the server again
 *
 * @api private
 */
MqttClient.prototype._checkPing = function () {
  if (this.pingResp) {
    this.pingResp = false;
    this.conn.pingreq();
  } else {
    // do a forced cleanup since socket will be in bad shape
    this._cleanUp(true);
  }
};

/**
 * _handlePingresp - handle a pingresp
 *
 * @api private
 */
MqttClient.prototype._handlePingresp = function () {
  this.pingResp = true;
};

/**
 * _handleConnack
 *
 * @param {Object} packet
 * @api private
 */

MqttClient.prototype._handleConnack = function(packet) {
  var rc = packet.returnCode;

  // TODO: move to protocol
  var errors = [
    '',
    'Unacceptable protocol version',
    'Identifier rejected',
    'Server unavailable',
    'Bad username or password',
    'Not authorized'
  ];

  if (rc === 0) {
    this.emit('connect');
  } else if (rc > 0) {
    this.emit('error',
        new Error('Connection refused: ' + errors[rc]));
  }
};

/**
 * _handlePublish
 *
 * @param {Object} packet
 * @api private
 */

MqttClient.prototype._handlePublish = function(packet) {
  var topic = packet.topic
    , message = packet.payload
    , qos = packet.qos
    , mid = packet.messageId
    , retain = packet.retain;

  switch (qos) {
    case 0:
      this.emit('message', topic, message, packet);
      break;
    case 1:
      this.conn.puback({messageId: mid});
      this.emit('message', topic, message, packet);
      break;
    case 2:
      this.conn.pubrec({messageId: mid});
      this.incoming.pubrel[mid] = packet;
      break;
    default:
      break;
  }
};

/**
 * _handleAck
 *
 * @param {Object} packet
 * @api private
 */

MqttClient.prototype._handleAck = function(packet) {
  var mid = packet.messageId
    , type = packet.cmd
    , cb = this.inflight[type][mid];

  if (!cb) {
    // Server sent an ack in error, ignore it.
    return;
  }

  // Process
  switch (type) {
    case 'puback':
      // Callback - we're done
      cb();
      break;
    case 'pubrec':
      // Pubrel and add to pubcomp list
      this.conn.pubrel(packet);
      this.inflight.pubcomp[mid] = cb;
      break;
    case 'pubcomp':
      // Callback - we're done
      cb();
      break;
    case 'suback':
      // TODO: RIDICULOUS HACK, PLEASE FIX
      var origSubs = cb.packet.subscriptions
        , cb = cb.callback
        , granted = packet.granted;

      for (var i = 0; i < granted.length; i += 1) {
        origSubs[i].qos = granted[i];
      }
      cb(null, origSubs);
      break;
    case 'unsuback':
      cb();
      break;
    default:
      // code
  }

  // Remove from queue
  delete this.inflight[type][mid];
};

/**
 * _handlePubrel
 *
 * @param {Object} packet
 * @api private
 */

MqttClient.prototype._handlePubrel = function(packet) {
  var mid = packet.messageId
    , pub = this.incoming.pubrel[mid];

  if (!pub) this.emit('error', new Error('Unknown message id'));
  this.conn.pubcomp({messageId: mid});
  this.emit('message', pub.topic, pub.payload, pub);
};

/**
 * _nextId
 */
MqttClient.prototype._nextId = function() {
  var id = this.nextId++;
  // Ensure 16 bit unsigned int:
  if (id === 65535) {
    this.nextId = 1;
  }
  return id;
};
