'use strict';

var util = require('util');
var transport = require('../spdy-transport');

var debug = {
  server: require('debug')('spdy:connection:server'),
  client: require('debug')('spdy:connection:client')
};
var EventEmitter = require('events').EventEmitter;

var Stream = transport.Stream;

function Connection(socket, options) {
  EventEmitter.call(this);

  var state = {};
  this._spdyState = state;

  // NOTE: There's a big trick here. Connection is used as a `this` argument
  // to the wrapped `connection` event listener.
  // socket end doesn't necessarly mean connection drop
  this.httpAllowHalfOpen = true;

  state.timeout = new transport.utils.Timeout(this);

  // Protocol info
  state.protocol = transport.protocol[options.protocol];
  state.version = null;
  state.constants = state.protocol.constants;
  state.pair = null;
  state.isServer = options.isServer;

  // Root of priority tree (i.e. stream id = 0)
  state.priorityRoot = new transport.Priority({
    defaultWeight: state.constants.DEFAULT_WEIGHT,
    maxCount: transport.protocol.base.constants.MAX_PRIORITY_STREAMS
  });

  // Defaults
  state.maxStreams = options.maxStreams ||
                     state.constants.MAX_CONCURRENT_STREAMS;

  state.autoSpdy31 = options.protocol.name !== 'h2' && options.autoSpdy31;
  state.acceptPush = options.acceptPush === undefined ?
      !state.isServer :
      options.acceptPush;

  if (options.maxChunk === false)
    state.maxChunk = Infinity;
  else if (options.maxChunk === undefined)
    state.maxChunk = transport.protocol.base.constants.DEFAULT_MAX_CHUNK;
  else
    state.maxChunk = options.maxChunk;

  // Connection-level flow control
  var windowSize = options.windowSize || 1 << 20;
  state.window = new transport.Window({
    id: 0,
    isServer: state.isServer,
    recv: {
      size: state.constants.DEFAULT_WINDOW,
      max: state.constants.MAX_INITIAL_WINDOW_SIZE
    },
    send: {
      size: state.constants.DEFAULT_WINDOW,
      max: state.constants.MAX_INITIAL_WINDOW_SIZE
    }
  });

  // It starts with DEFAULT_WINDOW, update must be sent to change it on client
  state.window.recv.setMax(windowSize);

  // Boilerplate for Stream constructor
  state.streamWindow = new transport.Window({
    id: -1,
    isServer: state.isServer,
    recv: {
      size: windowSize,
      max: state.constants.MAX_INITIAL_WINDOW_SIZE
    },
    send: {
      size: state.constants.DEFAULT_WINDOW,
      max: state.constants.MAX_INITIAL_WINDOW_SIZE
    }
  });

  // Various state info
  state.pool = state.protocol.compressionPool.create(options.headerCompression);
  state.counters = {
    push: 0,
    stream: 0
  };

  // Init streams list
  state.stream = {
    map: {},
    count: 0,
    nextId: state.isServer ? 2 : 1,
    lastId: {
      both: 0,
      received: 0
    }
  };
  state.ping = {
    nextId: state.isServer ? 2 : 1,
    map: {}
  };
  state.goaway = false;

  // Debug
  state.debug = state.isServer ? debug.server : debug.client;

  // X-Forwarded feature
  state.xForward = null;

  // Create parser and hole for framer
  state.parser = state.protocol.parser.create({
    // NOTE: needed to distinguish ping from ping ACK in SPDY
    isServer: state.isServer,
    window: state.window
  });
  state.framer = state.protocol.framer.create({
    window: state.window,
    timeout: state.timeout
  });

  // SPDY has PUSH enabled on servers
  if (state.protocol.name === 'spdy')
    state.framer.enablePush(state.isServer);

  if (!state.isServer)
    state.parser.skipPreface();

  this.socket = socket;

  this._init();
}
util.inherits(Connection, EventEmitter);
exports.Connection = Connection;

Connection.create = function create(socket, options) {
  return new Connection(socket, options);
};

Connection.prototype._init = function init() {
  var self = this;
  var state = this._spdyState;
  var pool = state.pool;

  // Initialize session window
  state.window.recv.on('drain', function() {
    self._onSessionWindowDrain();
  });

  // Initialize parser
  state.parser.on('data', function(frame) {
    self._handleFrame(frame);
  });
  state.parser.once('version', function(version) {
    self._onVersion(version);
  });

  // Propagate parser errors
  state.parser.on('error', function(err) {
    self._onParserError(err);
  });

  // Propagate framer errors
  state.framer.on('error', function(err) {
    self.emit('error', err);
  });

  this.socket.pipe(state.parser);
  state.framer.pipe(this.socket);

  // Allow high-level api to catch socket errors
  this.socket.on('error', function onSocketError(e) {
    self.emit('error', e);
  });

  this.socket.once('close', function onclose() {
    var err = new Error('socket hang up');
    err.code = 'ECONNRESET';
    self.destroyStreams(err);
    self.emit('close', err);

    if (state.pair)
      pool.put(state.pair);

    state.framer.resume();
  });

  // Reset timeout on close
  this.once('close', function() {
    self.setTimeout(0);
  });

  function _onWindowOverflow() {
    self._onWindowOverflow();
  }

  state.window.recv.on('overflow', _onWindowOverflow);
  state.window.send.on('overflow', _onWindowOverflow);

  // Do not allow half-open connections
  this.socket.allowHalfOpen = false;
};

Connection.prototype._onVersion = function _onVersion(version) {
  var state = this._spdyState;
  var prev = state.version;
  var parser = state.parser;
  var framer = state.framer;
  var pool = state.pool;

  state.version = version;
  state.debug('id=0 version=%d', version);

  // Ignore transition to 3.1
  if (!prev) {
    state.pair = pool.get(version);
    parser.setCompression(state.pair);
    framer.setCompression(state.pair);
  }
  framer.setVersion(version);

  if (!state.isServer) {
    framer.prefaceFrame();
    if (state.xForward !== null)
      framer.xForwardedFor({ host: state.xForward });
  }

  // Send preface+settings frame (once)
  framer.settingsFrame({
    max_header_list_size: state.constants.DEFAULT_MAX_HEADER_LIST_SIZE,
    max_concurrent_streams: state.maxStreams,
    enable_push: state.acceptPush ? 1 : 0,
    initial_window_size: state.window.recv.max
  });

  // Update session window
  if (state.version >= 3.1 || (state.isServer && state.autoSpdy31))
    this._onSessionWindowDrain();

  this.emit('version', version);
};

Connection.prototype._onParserError = function _onParserError(err) {
  var state = this._spdyState;

  // Prevent further errors
  state.parser.kill();

  // Send GOAWAY
  if (err instanceof transport.protocol.base.utils.ProtocolError) {
    this._goaway({
      lastId: state.stream.lastId.both,
      code: err.code,
      extra: err.message,
      send: true
    });
  }

  this.emit('error', err);
};

Connection.prototype._handleFrame = function _handleFrame(frame) {
  var state = this._spdyState;

  state.debug('id=0 frame', frame);
  state.timeout.reset();

  // For testing purposes
  this.emit('frame', frame);

  var stream;

  // Session window update
  if (frame.type === 'WINDOW_UPDATE' && frame.id === 0) {
    if (state.version < 3.1 && state.autoSpdy31) {
      state.debug('id=0 switch version to 3.1');
      state.version = 3.1;
      this.emit('version', 3.1);
    }
    state.window.send.update(frame.delta);
    return;
  }

  if (state.isServer && frame.type === 'PUSH_PROMISE') {
    state.debug('id=0 server PUSH_PROMISE');
    this._goaway({
      lastId: state.stream.lastId.both,
      code: 'PROTOCOL_ERROR',
      send: true
    });
    return;
  }

  if (!stream && frame.id !== undefined) {
    // Load created one
    stream = state.stream.map[frame.id];

    // Fail if not found
    if (!stream &&
        frame.type !== 'HEADERS' &&
        frame.type !== 'PRIORITY' &&
        frame.type !== 'RST') {
      // Other side should destroy the stream upon receiving GOAWAY
      if (this._isGoaway(frame.id))
        return;

      state.debug('id=0 stream=%d not found', frame.id);
      state.framer.rstFrame({ id: frame.id, code: 'INVALID_STREAM' });
      return;
    }
  }

  // Create new stream
  if (!stream && frame.type === 'HEADERS') {
    this._handleHeaders(frame);
    return;
  }

  if (stream) {
    stream._handleFrame(frame);
  } else if (frame.type === 'SETTINGS') {
    this._handleSettings(frame.settings);
  } else if (frame.type === 'ACK_SETTINGS') {
    // TODO(indutny): handle it one day
  } else if (frame.type === 'PING') {
    this._handlePing(frame);
  } else if (frame.type === 'GOAWAY') {
    this._handleGoaway(frame);
  } else if (frame.type === 'X_FORWARDED_FOR') {
    // Set X-Forwarded-For only once
    if (state.xForward === null)
      state.xForward = frame.host;
  } else if (frame.type === 'PRIORITY') {
    // TODO(indutny): handle this
  } else {
    state.debug('id=0 unknown frame type: %s', frame.type);
  }
};

Connection.prototype._onWindowOverflow = function _onWindowOverflow() {
  var state = this._spdyState;
  state.debug('id=0 window overflow');
  this._goaway({
    lastId: state.stream.lastId.both,
    code: 'FLOW_CONTROL_ERROR',
    send: true
  });
};

Connection.prototype._isGoaway = function _isGoaway(id) {
  var state = this._spdyState;
  if (state.goaway !== false && state.goaway < id)
    return true;
  return false;
};

Connection.prototype._getId = function _getId() {
  var state = this._spdyState;

  var id = state.stream.nextId;
  state.stream.nextId += 2;
  return id;
};

Connection.prototype._createStream = function _createStream(uri) {
  var state = this._spdyState;
  var id = uri.id;
  if (id === undefined)
    id = this._getId();

  var isGoaway = this._isGoaway(id);

  if (uri.push && !state.acceptPush) {
    state.debug('id=0 push disabled promisedId=%d', id);

    // Fatal error
    this._goaway({
      lastId: state.stream.lastId.both,
      code: 'PROTOCOL_ERROR',
      send: true
    });
    isGoaway = true;
  }

  var stream = new Stream(this, {
    id: id,
    request: uri.request !== false,
    method: uri.method,
    path: uri.path,
    host: uri.host,
    priority: uri.priority,
    headers: uri.headers,
    parent: uri.parent,
    readable: !isGoaway && uri.readable,
    writable: !isGoaway && uri.writable
  });
  var self = this;

  // Just an empty stream for API consistency
  if (isGoaway)
    return stream;

  state.stream.lastId.both = Math.max(state.stream.lastId.both, id);

  state.debug('id=0 add stream=%d', stream.id);
  state.stream.map[stream.id] = stream;
  state.stream.count++;
  state.counters.stream++;
  if (stream.parent !== null)
    state.counters.push++;

  stream.once('close', function() {
    self._removeStream(stream);
  });

  return stream;
};

Connection.prototype._handleHeaders = function _handleHeaders(frame) {
  var state = this._spdyState;

  // Must be HEADERS frame after stream close
  if (frame.id <= state.stream.lastId.received)
    return;

  // Someone is using our ids!
  if ((frame.id + state.stream.nextId) % 2 === 0) {
    state.framer.rstFrame({ id: frame.id, code: 'PROTOCOL_ERROR' });
    return;
  }

  var stream = this._createStream({
    id: frame.id,
    request: false,
    method: frame.headers[':method'],
    path: frame.headers[':path'],
    host: frame.headers[':authority'],
    priority: frame.priority,
    headers: frame.headers,
    writable: frame.writable
  });

  // GOAWAY
  if (this._isGoaway(stream.id))
    return;

  state.stream.lastId.received = Math.max(state.stream.lastId.received,
                                          stream.id);

  // TODO(indutny) handle stream limit
  if (!this.emit('stream', stream)) {
    // No listeners was set - abort the stream
    stream.abort();
    return;
  }

  // Create fake frame to simulate end of the data
  if (frame.fin)
    stream._handleFrame({ type: 'FIN', fin: true });

  return stream;
};

Connection.prototype._onSessionWindowDrain = function _onSessionWindowDrain() {
  var state = this._spdyState;
  if (state.version < 3.1 && !(state.isServer && state.autoSpdy31))
    return;

  var delta = state.window.recv.getDelta();
  if (delta === 0)
    return;

  state.debug('id=0 session window drain, update by %d', delta);

  state.framer.windowUpdateFrame({
    id: 0,
    delta: delta
  });
  state.window.recv.update(delta);
};

Connection.prototype.start = function start(version) {
  this._spdyState.parser.setVersion(version);
};

// Mostly for testing
Connection.prototype.getVersion = function getVersion() {
  return this._spdyState.version;
};

Connection.prototype._handleSettings = function _handleSettings(settings) {
  var state = this._spdyState;

  state.framer.ackSettingsFrame();

  this._setDefaultWindow(settings);
  if (settings.max_frame_size)
    state.framer.setMaxFrameSize(settings.max_frame_size);

  // TODO(indutny): handle max_header_list_size
  if (settings.header_table_size) {
    try {
      state.pair.compress.updateTableSize(settings.header_table_size);
    } catch (e) {
      this._goaway({
        lastId: 0,
        code: 'PROTOCOL_ERROR',
        send: true
      });
      return;
    }
  }

  // HTTP2 clients needs to enable PUSH streams explicitly
  if (state.protocol.name !== 'spdy') {
    if (settings.enable_push === undefined)
      state.framer.enablePush(state.isServer);
    else
      state.framer.enablePush(settings.enable_push === 1);
  }

  // TODO(indutny): handle max_concurrent_streams
};

Connection.prototype._setDefaultWindow = function _setDefaultWindow(settings) {
  if (settings.initial_window_size === undefined)
    return;

  var state = this._spdyState;

  // Update defaults
  var window = state.streamWindow;
  window.send.setMax(settings.initial_window_size);

  // Update existing streams
  Object.keys(state.stream.map).forEach(function(id) {
    var stream = state.stream.map[id];
    var window = stream._spdyState.window;

    window.send.updateMax(settings.initial_window_size);
  });
};

Connection.prototype._handlePing = function handlePing(frame) {
  var self = this;
  var state = this._spdyState;

  // Handle incoming PING
  if (!frame.ack) {
    state.framer.pingFrame({
      opaque: frame.opaque,
      ack: true
    });

    self.emit('ping', frame.opaque);
    return;
  }

  // Handle reply PING
  var hex = frame.opaque.toString('hex');
  if (!state.ping.map[hex])
    return;
  var ping = state.ping.map[hex];
  delete state.ping.map[hex];

  if (ping.cb)
    ping.cb(null);
};

Connection.prototype._handleGoaway = function handleGoaway(frame) {
  this._goaway({
    lastId: frame.lastId,
    code: frame.code,
    send: false
  });
};

Connection.prototype.ping = function ping(callback) {
  var state = this._spdyState;

  // HTTP2 is using 8-byte opaque
  var opaque = new Buffer(state.constants.PING_OPAQUE_SIZE);
  opaque.fill(0);
  opaque.writeUInt32BE(state.ping.nextId, opaque.length - 4);
  state.ping.nextId += 2;

  state.ping.map[opaque.toString('hex')] = { cb: callback };
  state.framer.pingFrame({
    opaque: opaque,
    ack: false
  });
};

Connection.prototype.getCounter = function getCounter(name) {
  return this._spdyState.counters[name];
};

Connection.prototype.reserveStream = function reserveStream(uri, callback) {
  var stream = this._createStream(uri);

  // GOAWAY
  if (this._isGoaway(stream.id)) {
    var err = new Error('Can\'t send request after GOAWAY');
    process.nextTick(function() {
      if (callback)
        callback(err);
      else
        stream.emit('error', err);
    });
    return stream;
  }

  if (callback) {
    process.nextTick(function() {
      callback(null, stream);
    });
  }

  return stream;
};

Connection.prototype.request = function request(uri, callback) {
  var stream = this.reserveStream(uri, function(err) {
    if (err) {
      if (callback)
        callback(err);
      else
        stream.emit('error', err);
      return;
    }

    if (stream._wasSent()) {
      if (callback)
        callback(null, stream);
      return;
    }

    stream.send(function(err) {
      if (err) {
        if (callback)
          return callback(err);
        else
          return stream.emit('error', err);
      }

      if (callback)
        callback(null, stream);
    });
  });

  return stream;
};

Connection.prototype._removeStream = function _removeStream(stream) {
  var state = this._spdyState;

  state.debug('id=0 remove stream=%d', stream.id);
  delete state.stream.map[stream.id];
  state.stream.count--;

  if (state.stream.count === 0)
    this.emit('_streamDrain');
};

Connection.prototype._goaway = function _goaway(params) {
  var state = this._spdyState;
  var self = this;

  state.goaway = params.lastId;
  state.debug('id=0 goaway from=%d', state.goaway);

  Object.keys(state.stream.map).forEach(function(id) {
    var stream = state.stream.map[id];

    // Abort every stream started after GOAWAY
    if (stream.id <= params.lastId)
      return;

    stream.abort();
    stream.emit('error', new Error('New stream after GOAWAY'));
  });

  function finish() {
    // Destroy socket if there are no streams
    if (state.stream.count === 0 || params.code !== 'OK') {
      // No further frames should be processed
      state.parser.kill();

      process.nextTick(function() {
        var err = new Error('Fatal error: ' + params.code);
        self._onStreamDrain(err);
      });
      return;
    }

    self.on('_streamDrain', self._onStreamDrain);
  }

  if (params.send) {
    // Make sure that GOAWAY frame is sent before dumping framer
    state.framer.goawayFrame({
      lastId: params.lastId,
      code: params.code,
      extra: params.extra
    }, finish);
  } else {
    finish();
  }
};

Connection.prototype._onStreamDrain = function _onStreamDrain(error) {
  var state = this._spdyState;

  state.debug('id=0 _onStreamDrain');

  state.framer.dump();
  state.framer.unpipe(this.socket);
  state.framer.resume();

  if (this.socket.destroySoon)
    this.socket.destroySoon();
  this.emit('close', error);
};

Connection.prototype.end = function end(callback) {
  var state = this._spdyState;

  if (callback)
    this.once('close', callback);
  this._goaway({
    lastId: state.stream.lastId.both,
    code: 'OK',
    send: true
  });
};

Connection.prototype.destroyStreams = function destroyStreams(err) {
  var state = this._spdyState;
  Object.keys(state.stream.map).forEach(function(id) {
    var stream = state.stream.map[id];

    stream.abort();
    stream.emit('error', err);
  });
};

Connection.prototype.isServer = function isServer() {
  return this._spdyState.isServer;
};

Connection.prototype.getXForwardedFor = function getXForwardFor() {
  return this._spdyState.xForward;
};

Connection.prototype.sendXForwardedFor = function sendXForwardedFor(host) {
  var state = this._spdyState;
  if (state.version !== null)
    state.framer.xForwardedFor({ host: host });
  else
    state.xForward = host;
};

Connection.prototype.pushPromise = function pushPromise(parent, uri, callback) {
  var state = this._spdyState;

  var stream = this._createStream({
    request: false,
    parent: parent,
    method: uri.method,
    path: uri.path,
    host: uri.host,
    priority: uri.priority,
    headers: uri.headers,
    readable: false
  });

  // TODO(indutny): deduplicate this logic somehow
  if (this._isGoaway(stream.id)) {
    var err = new Error('Can\'t send PUSH_PROMISE after GOAWAY');
    process.nextTick(function() {
      if (callback)
        callback(err);
      else
        stream.emit('error', err);
    });
    return stream;
  }

  if (uri.push && !state.acceptPush) {
    var err = new Error(
        'Can\'t send PUSH_PROMISE, other side won\'t accept it');
    process.nextTick(function() {
      if (callback)
        callback(err);
      else
        stream.emit('error', err);
    });
    return stream;
  }

  stream._sendPush(uri.status, uri.response, function(err) {
    if (!callback) {
      if (err)
        stream.emit('error', err);
      return;
    }

    if (err)
      return callback(err);
    callback(null, stream);
  });

  return stream;
};

Connection.prototype.setTimeout = function setTimeout(delay, callback) {
  var state = this._spdyState;

  state.timeout.set(delay, callback);
};
