// -*- mode: c++; c-basic-offset: 2; -*-

#include "common.hh"

#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif

#include <arpa/inet.h>
#include <cerrno>
#include <chrono>
#include <cstring>
#include <fcntl.h>
#include <memory>
#include <netdb.h>
#include <signal.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
#include <vector>
#include <unordered_set>

#include "buffer.hh"
#include "chunked.hh"
#include "config.hh"
#include "data.hh"
#include "http.hh"
#include "io.hh"
#include "logger.hh"
#include "looper.hh"
#include "mitm.hh"
#include "resolver.hh"
#include "package.hh"
#include "paths.hh"
#include "proxy.hh"
#include "url.hh"

namespace {

auto const NEW_CONNECTION_TIMEOUT = std::chrono::duration<float>(5.0f);
auto const CONNECTION_TIMEOUT = std::chrono::duration<float>(30.0f);

template<typename T>
class clients {
public:
  class iterator {
  public:
    iterator(clients* clients, size_t active)
      : clients_(clients), index_(0), active_(active) {
      if (active_) {
        while (clients_->is_free(index_)) {
          ++index_;
        }
      }
    }
    iterator(iterator const& i)
      : clients_(i.clients_), index_(i.index_), active_(i.active_) {
    }

    iterator& operator=(iterator const& i) {
      clients_ = i.clients_;
      index_ = i.index_;
      active_ = i.active_;
      return *this;
    }

    bool operator==(iterator const& i) const {
      return active_ == i.active_;
    }
    bool operator!=(iterator const& i) const {
      return !(*this == i);
    }
    bool operator<(iterator const& i) const {
      return active_ < i.active_;
    }
    bool operator<=(iterator const& i) const {
      return (*this < i) || (*this == i);
    }
    bool operator>=(iterator const& i) const {
      return !(*this < i);
    }
    bool operator>(iterator const& i) const {
      return !(*this <= i);
    }

    iterator& operator++() {
      next();
      return *this;
    }

    iterator operator++(int UNUSED(dummy)) {
      iterator ret(*this);
      ++(*this);
      return ret;
    }

    T& operator*() {
      return (*clients_)[index_];
    }
    T* operator->() {
      return &(*clients_)[index_];
    }

    size_t index() const {
      return index_;
    }

  private:
    void next() {
      if (active_ == 0) return;
      --active_;
      if (active_ == 0) return;
      ++index_;
      while (clients_->is_free(index_)) {
        ++index_;
      }
    }

    clients* clients_;
    size_t index_;
    size_t active_;
  };

  clients()
    : active_(0), max_(0) {
  }

  void resize(size_t max) {
    max_ = max;
    if (max > data_.size()) {
      data_.resize(max);
    }
  }

  bool full() const {
    return active_ >= max_;
  }

  iterator begin() {
    return iterator(this, active_);
  }

  iterator end() {
    return iterator(this, 0);
  }

  T& operator[] (size_t index) {
    return data_[index];
  }

  size_t new_client() {
    size_t index;
    if (active_ == data_.size()) {
      index = data_.size();
      data_.emplace_back();
    } else {
      index = rand() % data_.size();
      while (!is_free(index)) {
        if (++index == data_.size()) {
          index = 0;
        }
      }
    }
    ++active_;
    return index;
  }

  void erase(size_t index) {
    assert(active_ > 0);
    --active_;
    assert(is_free(index));
    if (data_.size() > max_ && index >= max_) {
      size_t size = data_.size();
      while (size > max_ && is_free(size - 1)) --size;
      data_.resize(size);
    }
  }

  void clear() {
    if (active_ > 0) {
      data_.clear();
      active_ = 0;
    }
    data_.resize(max_);
  }

private:
  bool is_free(size_t index) const {
    return !data_[index].fd;
  }

  size_t active_;
  size_t max_;
  std::vector<T> data_;
};

struct BaseClient {
  io::auto_fd fd;
  bool new_connection;
  Looper::clock::time_point last;
  std::unique_ptr<Buffer> in;
  std::unique_ptr<Buffer> out;
  uint8_t read_flag;
  uint8_t write_flag;
};

enum ContentType {
  CONTENT_NONE,
  CONTENT_LEN,
  CONTENT_CHUNKED,
  CONTENT_CLOSE
};

struct Content {
  ContentType type;
  uint64_t len;
  std::unique_ptr<Chunked> chunked;
};

struct Connect {
  std::string host;
  uint16_t port;
  Mitm::DetectResult mitm_detect;
  std::unique_ptr<Mitm::Connection> mitm;
  std::unique_ptr<Mitm::Monitor> mitm_monitor;
};

enum RemoteState {
  CLOSED,
  RESOLVING,
  CONNECTING,
  CONNECTED,
  WAITING,
};

struct RemoteClient : public BaseClient {
  Content content;
  std::string host;
  uint16_t port;
  uint32_t pkg_id;
};

struct Client : public BaseClient {
  Client()
    : resolve(nullptr) {
  }
  std::unique_ptr<HttpRequest> request;
  std::unique_ptr<Connect> connect;
  std::unique_ptr<Url> url;
  Content content;
  RemoteState remote_state;
  void* resolve;
  RemoteClient remote;
  std::string source_host;
  uint16_t source_port;
  uint32_t pkg_id;
};

struct Monitor : public BaseClient {
  bool got_hello;
};

class ProxyImpl : public Proxy {
public:
  ProxyImpl(Config* config, std::string const& cwd, char const* configfile,
            char const* logfile, Logger* logger, int accept_fd, int monitor_fd,
            Looper* looper, Resolver* resolver)
    : config_(config), cwd_(cwd), configfile_(configfile), logfile_(logfile),
      logger_(logger), accept_socket_(accept_fd), monitor_socket_(monitor_fd),
      looper_(looper), resolver_(resolver),
      new_timeout_(nullptr), timeout_(nullptr), next_package_id_(1),
      monitor_send_proxied_(false) {
    setup();
  }
  ~ProxyImpl() override {
    while (true) {
      auto it = clients_.begin();
      if (it == clients_.end()) break;
      close_client(it.index());
    }
    while (true) {
      auto it = monitors_.begin();
      if (it == monitors_.end()) break;
      close_monitor(it.index());
    }

    if (timeout_) {
      looper_->cancel(timeout_);
    }
    if (new_timeout_) {
      looper_->cancel(new_timeout_);
    }

    if (accept_socket_) {
      looper_->remove(accept_socket_.get());
    }
    if (monitor_socket_) {
      looper_->remove(monitor_socket_.get());
    }
    if (signal_pipe_) {
      looper_->remove(signal_pipe_.read());
    }
  }

  void quit() override {
    char cmd = 'q';
    io::write_all(signal_pipe_.write(), &cmd, 1);
  }

  void reload() override {
    char cmd = 'r';
    io::write_all(signal_pipe_.write(), &cmd, 1);
  }

  bool run() override;

private:
  class MitmMonitor : public Mitm::Monitor {
  public:
    MitmMonitor(ProxyImpl* proxy, uint32_t local_pkg_id, uint32_t remote_pkg_id)
      : proxy_(proxy), local_pkg_id_(local_pkg_id),
        remote_pkg_id_(remote_pkg_id) {
      assert(local_pkg_id_ && remote_pkg_id_);
    }

    void local2remote(void const* data, size_t size) override {
      proxy_->send_attached_data(local_pkg_id_, data, size, false);
    }
    void remote2local(void const* data, size_t size) override {
      proxy_->send_attached_data(remote_pkg_id_, data, size, false);
    }

  private:
    ProxyImpl* const proxy_;
    uint32_t const local_pkg_id_;
    uint32_t const remote_pkg_id_;
  };

  void setup();
  bool reload_config();
  void fatal_error();
  void new_client(int fd, uint8_t events);
  void new_monitor(int fd, uint8_t events);
  void setup_monitor(int fd);
  void new_base(BaseClient* client, int fd);
  void signal_event(int fd, uint8_t events);
  bool base_event(BaseClient* client, uint8_t events,
                  size_t index, char const* name);
  void client_event(size_t index, int fd, uint8_t events);
  void client_remote_event(size_t index, int fd, uint8_t events);
  void client_empty_input(size_t index);
  void monitor_event(size_t index, int fd, uint8_t events);
  void close_client(size_t index);
  void close_monitor(size_t index);
  void close_base(BaseClient* client);
  void new_timeout();
  void timeout();
  float handle_timeout(bool new_conn,
                       std::chrono::duration<float> const& timeout);
  bool base_send(BaseClient* client, void const* data, size_t size,
                  size_t index, char const* name);
  bool base_continue_send(BaseClient* client, bool was_empty,
                          size_t index, char const* name);
  void client_error(size_t index,
                    uint16_t status_code, std::string const& status);
  bool client_request(size_t index);
  void client_remote_error(size_t index, uint16_t error);
  void close_client_when_done(size_t index);
  void client_remote_resolved(size_t index, int fd, bool connected,
                              char const* error);
  void monitor_error(size_t index,
                     uint16_t status_code, std::string const& status);
  bool support_monitor_version(size_t index, std::string const& version);
  bool monitor_send_chunked(size_t index,
                            void const* header, size_t header_size,
                            void const* data, size_t data_size);
  uint32_t get_next_package_id();
  void send_attached_package(uint32_t id, uint16_t flags,
                             std::string const& source_host,
                             uint16_t source_port,
                             std::string const& target_host,
                             uint16_t target_port,
                             bool last);
  void send_attached_data(uint32_t id, void const* ptr, size_t size, bool last);
  void send_attached(void const* header, size_t header_size,
                     void const* data, size_t data_size);

  Config* const config_;
  std::string cwd_;
  char const* const configfile_;
  char const* const logfile_;
  Logger* logger_;
  std::unique_ptr<Logger> priv_logger_;
  io::auto_fd accept_socket_;
  io::auto_fd monitor_socket_;
  io::auto_pipe signal_pipe_;
  Looper* looper_;
  Resolver* resolver_;
  std::unique_ptr<Mitm> mitm_;
  bool good_;
  void* new_timeout_;
  void* timeout_;
  uint32_t next_package_id_;
  bool monitor_send_proxied_;

  clients<Client> clients_;
  clients<Monitor> monitors_;
  std::unordered_set<size_t> attached_;
};

size_t get_size(Config* config, Logger* logger, std::string const& name,
                size_t fallback) {
  auto value = config->get(name, nullptr);
  if (!value) return fallback;
  char* end = nullptr;
  errno = 0;
  auto tmp = strtoul(value, &end, 10);
  if (errno || !end || *end) {
    logger->out(Logger::WARN,
                "Invalid value given for %s: %s, using fallback %zu instead",
                name.c_str(), value, fallback);
    return fallback;
  }
  return static_cast<size_t>(tmp);
}

void ProxyImpl::setup() {
  monitor_send_proxied_ = !config_->get("monitor_proxy_request", false);
  clients_.resize(get_size(config_, logger_, "max_clients", 1024));
  monitors_.resize(get_size(config_, logger_, "max_monitors", 2));
  mitm_.reset(Mitm::create(logger_, config_, cwd_));
  if (mitm_) {
    logger_->out(Logger::INFO, "MITM SSL Interception active");
  }
  looper_->add(accept_socket_.get(),
               clients_.full() ? 0 : Looper::EVENT_READ,
               std::bind(&ProxyImpl::new_client, this,
                         std::placeholders::_1,
                         std::placeholders::_2));
  if (monitor_socket_) {
    if (config_->get("__one_single_monitor", false)) {
      setup_monitor(monitor_socket_.release());
    } else {
      looper_->add(monitor_socket_.get(),
                   monitors_.full() ? 0 :Looper::EVENT_READ,
                   std::bind(&ProxyImpl::new_monitor, this,
                             std::placeholders::_1,
                             std::placeholders::_2));
    }
  } else {
    monitors_.clear();
  }
}

bool ProxyImpl::reload_config() {
  if (configfile_) {
    auto file = Paths::join(cwd_, configfile_);
    logger_->out(Logger::INFO, "Reloading config file %s", file.c_str());
    config_->load_file(file);
  } else {
    logger_->out(Logger::INFO, "Reloading config");
    config_->load_name("tp");
  }
  if (!config_->good()) {
    logger_->out(Logger::WARN, "New config invalid, ignored.");
    return true;
  }
  monitor_send_proxied_ = !config_->get("monitor_proxy_request", false);
  if (!logfile_) {
    auto logfile = config_->get("logfile", nullptr);
    if (logfile) {
      if (logfile[0] != '/') {
        logger_->out(Logger::WARN,
                     "Logfile need to be an absolute path, not: %s",
                     logfile);
      } else {
        std::unique_ptr<Logger> tmp(Logger::create_file(logfile_));
        if (tmp) {
          priv_logger_.swap(tmp);
          logger_ = priv_logger_.get();
        } else {
          logger_->out(Logger::WARN,
                       "Unable to open %s for logging", logfile);
        }
      }
    } else {
      priv_logger_.reset(Logger::create_syslog("tp"));
      logger_ = priv_logger_.get();
    }
  }
  if (mitm_) {
    if (!mitm_->reload(config_, cwd_)) {
      logger_->out(Logger::WARN, "Invalid mitm config, ignored");
    }
  } else {
    mitm_.reset(Mitm::create(logger_, config_, cwd_));
  }
  auto const old_accept = accept_socket_.get();
  auto const old_monitor = monitor_socket_.get();
  looper_->remove(old_accept);
  if (monitor_socket_) looper_->remove(old_monitor);
  accept_socket_.reset(setup_accept_socket(config_, logger_));
  monitor_socket_.reset();
  if (!accept_socket_) {
    logger_->out(Logger::ERR, "Unable to bind to new configuration, abort");
    return false;
  }
  if (config_->get("monitor", false)) {
    monitor_socket_.reset(setup_monitor_socket(config_, logger_));
    if (!monitor_socket_) {
      logger_->out(Logger::ERR, "Unable to bind to new configuration, abort");
      return false;
    }
  }
  setup();
  return true;
}

void ProxyImpl::signal_event(int fd, uint8_t events) {
  assert(fd == signal_pipe_.read());
  if (events == Looper::EVENT_READ) {
    char cmd;
    auto ret = io::read(fd, &cmd, 1);
    if (ret == 1) {
      switch (cmd) {
      case 'q':
        logger_->out(Logger::INFO, "Exiting ...");
        looper_->quit();
        return;
      case 'r':
        if (!reload_config()) {
          fatal_error();
          return;
        }
        break;
      }
    }
  } else {
    logger_->out(Logger::WARN, "Signal pipes have crashed");
    signal_pipe_.reset();
    looper_->remove(fd);
  }
}

void ProxyImpl::close_base(BaseClient* client) {
  if (client->fd) {
    looper_->remove(client->fd.get());
    client->fd.reset();
  }
  client->in.reset();
  client->out.reset();
}

void ProxyImpl::close_client(size_t index) {
  bool was_full = clients_.full();
  auto& client = clients_[index];
  if (client.pkg_id != 0) {
    size_t avail;
    auto ptr = client.in->read_ptr(&avail);
    if (avail) {
      send_attached_data(client.pkg_id, ptr, avail, true);
    } else {
      send_attached_data(client.pkg_id, nullptr, 0, true);
    }
    client.pkg_id = 0;
  }
  client.request.reset();
  client.url.reset();
  client.connect.reset();
  client.content.type = CONTENT_NONE;
  client.content.chunked.reset();
  client.remote_state = CLOSED;
  if (client.resolve) {
    resolver_->cancel(client.resolve);
    client.resolve = nullptr;
  }
  if (client.remote.pkg_id) {
    size_t avail;
    auto ptr = client.remote.in->read_ptr(&avail);
    if (avail) {
      send_attached_data(client.remote.pkg_id, ptr, avail, true);
    } else {
      send_attached_data(client.remote.pkg_id, nullptr, 0, true);
    }
    client.remote.pkg_id = 0;
  }
  close_base(&client.remote);
  close_base(&client);
  clients_.erase(index);
  if (was_full && !clients_.full() && accept_socket_) {
    looper_->modify(accept_socket_.get(), Looper::EVENT_READ);
  }
}

void ProxyImpl::close_monitor(size_t index) {
  bool was_full = monitors_.full();
  auto& monitor = monitors_[index];
  auto it = attached_.find(index);
  if (it != attached_.end()) {
    attached_.erase(it);
  }
  close_base(&monitor);
  monitors_.erase(index);
  if (was_full && !monitors_.full() && monitor_socket_) {
    looper_->modify(monitor_socket_.get(), Looper::EVENT_READ);
  }
}

float ProxyImpl::handle_timeout(bool new_conn,
                                std::chrono::duration<float> const& timeout) {
  auto now = looper_->now();
  std::vector<size_t> close;
  std::vector<bool> remote;
  float next = -1.0f;
  for (auto i = clients_.begin(); i != clients_.end(); ++i) {
    if (i->new_connection != new_conn) continue;
    auto diff = std::chrono::duration_cast<std::chrono::duration<float>>(
        ((i->last + timeout) - now)).count();
    if (diff < 0.0f) {
      close.push_back(i.index());
      remote.push_back(false);
    } else {
      if (!new_conn && i->remote_state > CLOSED) {
        auto diff2 = std::chrono::duration_cast<std::chrono::duration<float>>(
            ((i->remote.last + timeout) - now)).count();
        if (diff2 < 0.0f) {
          close.push_back(i.index());
          remote.push_back(true);
        } else if (diff2 < diff) {
          diff = diff2;
        }
      }
      if (next < 0.0f || diff < next) {
        next = diff;
      }
    }
  }
  assert(close.size() == remote.size());
  auto j = remote.rbegin();
  for (auto i = close.rbegin(); i != close.rend(); ++i, ++j) {
    if (*j) {
      client_remote_error(*i, 504);
    } else {
      close_client(*i);
    }
  }
  close.clear();
  for (auto i = monitors_.begin(); i != monitors_.end(); ++i) {
    if (i->new_connection != new_conn) continue;
    // Monitors are safe from timeout after hello
    if (i->got_hello) continue;
    auto diff = std::chrono::duration_cast<std::chrono::duration<float>>(
        ((i->last + timeout) - now)).count();
    if (diff < 0.0f) {
      close.push_back(i.index());
    } else if (next < 0.0f || diff < next) {
      next = diff;
    }
  }
  for (auto i = close.rbegin(); i != close.rend(); ++i) {
    close_monitor(*i);
  }
  return next;
}

void ProxyImpl::timeout() {
  assert(timeout_);
  timeout_ = nullptr;
  float next = handle_timeout(false, CONNECTION_TIMEOUT);
  if (next < 0.0f) return;
  timeout_ = looper_->schedule(next, std::bind(&ProxyImpl::timeout, this));
}

void ProxyImpl::new_timeout() {
  assert(new_timeout_);
  new_timeout_ = nullptr;
  float next = handle_timeout(true, NEW_CONNECTION_TIMEOUT);
  if (next < 0.0f) return;
  new_timeout_ =
    looper_->schedule(next, std::bind(&ProxyImpl::new_timeout, this));
}

bool ProxyImpl::base_send(BaseClient* client, void const* data, size_t size,
                          size_t index, char const* name) {
  if (size == 0) return true;
  if (!client->out->empty()) {
    // Already waiting for write event
    client->out->write(data, size);
    return true;
  }
  auto ret = io::write(client->fd.get(), data, size);
  if (ret == -1) {
    if (errno != EAGAIN && errno != EWOULDBLOCK) {
      logger_->out(Logger::INFO, "%zu: %s write error: %s", index, name,
                   strerror(errno));
      return false;
    }
    client->out->write(data, size);
  } else {
    if (static_cast<size_t>(ret) == size) {
      if (!client->in) {
        // If input is closed, close after sending all data
        return false;
      }
      return true;
    }
    client->out->write(reinterpret_cast<char const*>(data) + ret, size - ret);
  }
  client->write_flag = Looper::EVENT_WRITE;
  looper_->modify(client->fd.get(), client->read_flag | client->write_flag);
  return true;
}

bool ProxyImpl::base_continue_send(BaseClient* client, bool was_empty,
                                   size_t index, char const* name) {
  if (client->out->empty()) return true;
  if (!was_empty) {
    // Already waiting for write event
    return true;
  }
  size_t avail;
  auto ptr = client->out->read_ptr(&avail);
  auto ret = io::write(client->fd.get(), ptr, avail);
  if (ret == -1) {
    if (errno != EAGAIN && errno != EWOULDBLOCK) {
      logger_->out(Logger::INFO, "%zu: %s write error: %s", index, name,
                   strerror(errno));
      return false;
    }
  } else {
    client->out->consume(ret);
    if (static_cast<size_t>(ret) == avail) {
      if (!client->in) {
        // If input is closed, close after sending all data
        return false;
      }
      return true;
    }
  }
  client->write_flag = Looper::EVENT_WRITE;
  looper_->modify(client->fd.get(), client->read_flag | client->write_flag);
  return true;
}

bool ProxyImpl::base_event(BaseClient* client, uint8_t events,
                           size_t index, char const* name) {
  if (events & Looper::EVENT_READ) {
    if (client->new_connection) {
      char tmp[1];
      auto ret = io::read(client->fd.get(), &tmp, 1);
      if (ret < 0) {
        if (errno == EAGAIN || errno == EWOULDBLOCK) return true;
        logger_->out(Logger::INFO, "%zu: %s read error: %s", index, name,
                     strerror(errno));
        return false;
      }
      if (ret == 0) {
        return false;
      }
      client->last = looper_->now();
      client->new_connection = false;
      client->in.reset(Buffer::create(8192, 1024));
      client->out.reset(Buffer::create(8192, 1024));
      client->in->write(tmp, 1);
      if (!timeout_) {
        timeout_ = looper_->schedule(CONNECTION_TIMEOUT.count(),
                                     std::bind(&ProxyImpl::timeout, this));
      }
    }
    if (!client->in) {
      assert(false);
      return false;
    }
    size_t avail;
    auto ptr = client->in->write_ptr(&avail);
    assert(avail > 0);
    auto ret = io::read(client->fd.get(), ptr, avail);
    if (ret < 0) {
      if (errno == EAGAIN || errno == EWOULDBLOCK) return true;
      logger_->out(Logger::INFO, "%zu: %s read error: %s", index, name,
                   strerror(errno));
      return false;
    }
    if (ret == 0) {
      return false;
    }
    client->last = looper_->now();
    client->in->commit(ret);
  }
  if (events & Looper::EVENT_WRITE) {
    if (client->new_connection) {
      assert(false);
      return true;
    }
    size_t avail;
    auto ptr = client->out->read_ptr(&avail);
    if (avail == 0) {
      assert(false);
      if (!client->in) return false;
      client->write_flag = 0;
      looper_->modify(client->fd.get(), client->read_flag);
      return true;
    }
    auto ret = io::write(client->fd.get(), ptr, avail);
    if (ret < 0) {
      if (errno == EAGAIN || errno == EWOULDBLOCK) return true;
      logger_->out(Logger::INFO, "%zu: %s write error: %s", index, name,
                   strerror(errno));
      return false;
    }
    if (ret == 0) {
      return false;
    }
    client->last = looper_->now();
    client->out->consume(ret);
    if (client->out->empty()) {
      if (!client->in) return false;
      client->write_flag = 0;
      looper_->modify(client->fd.get(), client->read_flag);
    }
  }
  return true;
}

inline char lower_ascii(char c) {
  return (c >= 'A' && c <= 'Z') ? (c - 'A' + 'a') : c;
}

bool lower_equal(char const* data, size_t start, size_t end,
                 std::string const& str) {
  assert(start <= end);
  if (str.size() != end - start) return false;
  for (auto i = str.begin(); start < end; ++start, ++i) {
    if (lower_ascii(*i) != lower_ascii(data[start])) return false;
  }
  return true;
}

bool header_token_eq(std::string const& value, std::string const& token) {
  if (value.empty()) return false;
  auto pos = value.find(';');
  if (pos == std::string::npos) pos = value.size();
  return lower_equal(value.data(), 0, pos, token);
}

void ProxyImpl::client_remote_error(size_t index, uint16_t error) {
  auto& client = clients_[index];
  if (client.remote_state > CONNECTED) {
    // Already started to return response, too late to do anything
    close_client(index);
    return;
  }
  char const* status;
  switch (error) {
  case 502:
    status = "Bad Gateway";
    break;
  case 504:
    status = "Gateway Timeout";
    break;
  default:
    assert(false);
    error = 500;
    status = "Internal Server Error";
    break;
  }

  client_error(index, error, status);
}

void ProxyImpl::client_error(size_t index, uint16_t status_code,
                             std::string const& status_message) {
  auto& client = clients_[index];
  // No more input
  client.read_flag = 0;
  looper_->modify(client.fd.get(), client.write_flag);
  client.url.reset();
  client.connect.reset();

  auto resp = std::unique_ptr<HttpResponseBuilder>(
      HttpResponseBuilder::create(
          "HTTP", Version(1, 0), status_code, status_message));
  resp->add_header("Content-Length", "0");
  resp->add_header("Connection", "close");

  client.in.reset();
  client.request.reset();

  auto data = resp->build();

  if (!base_send(&client, data.data(), data.size(), index, "Client")) {
    close_client(index);
    return;
  }
}

bool ProxyImpl::client_request(size_t index) {
  auto& client = clients_[index];
  auto version = client.request->proto_version();
  if (version.major != 1 || version.minor > 1) {
    client_error(index, 505, "HTTP Version Not Supported");
    return false;
  }
  std::string host;
  uint16_t port;
  if (client.connect) {
    host = client.connect->host;
    port = client.connect->port;
  } else {
    if (client.url->scheme() != "http") {
      client_error(index, 501, "Not Implemented");
      return false;
    }
    if (client.url->userinfo()) {
      client_error(index, 400, "Bad request");
      return false;
    }
    host = client.url->host();
    port = client.url->port();
  }
  if (port == 0) port = 80;
  client.pkg_id = get_next_package_id();
  if (client.pkg_id != 0) {
    send_attached_package(client.pkg_id, 0,
                          client.source_host, client.source_port,
                          host, port, false);
  }
  if (client.remote_state == WAITING) {
    if (client.connect || host != client.remote.host
            || port != client.remote.port) {
      client.remote_state = CLOSED;
      close_base(&client.remote);
    } else {
      client.remote_state = CONNECTING;
      client.remote.read_flag = Looper::EVENT_READ;
      client.remote.write_flag = 0;
      looper_->modify(client.remote.fd.get(),
                      client.remote.read_flag | client.remote.write_flag);
      client_remote_event(index, client.remote.fd.get(), Looper::EVENT_WRITE);
      return true;
    }
  }
  assert(client.remote_state == CLOSED);
  client.remote.last = looper_->now();
  client.remote.new_connection = true;
  client.remote_state = RESOLVING;
  client.remote.host = host;
  client.remote.port = port;

  client.resolve = resolver_->request(
      host, port,
      std::bind(&ProxyImpl::client_remote_resolved, this, index,
                std::placeholders::_1,
                std::placeholders::_2,
                std::placeholders::_3));
  return true;
}

void ProxyImpl::client_remote_resolved(size_t index, int fd, bool connected,
                                       char const* error) {
  auto& client = clients_[index];
  assert(client.resolve);
  client.resolve = nullptr;
  assert(client.remote_state == RESOLVING);
  if (fd < 0) {
    logger_->out(Logger::INFO, "%zu: Client unable to resolve remote: %s",
                 index, error);
    client_remote_error(index, 502);
    return;
  }
  client.remote_state = CONNECTING;
  client.remote.fd.reset(fd);
  client.remote.last = looper_->now();
  client.remote.new_connection = false;
  client.remote.in.reset(Buffer::create(8192, 1024));
  client.remote.out.reset(Buffer::create(8192, 1024));
  if (connected) {
    client.remote.read_flag = Looper::EVENT_READ;
    client.remote.write_flag = 0;
  } else {
    client.remote.read_flag = 0;
    client.remote.write_flag = Looper::EVENT_WRITE;
  }
  looper_->add(client.remote.fd.get(),
               client.remote.read_flag | client.remote.write_flag,
               std::bind(&ProxyImpl::client_remote_event, this, index,
                         std::placeholders::_1,
                         std::placeholders::_2));
  assert(timeout_);
  if (connected) {
    client_remote_event(index, fd, Looper::EVENT_WRITE);
  }
}

void ProxyImpl::client_event(size_t index, int fd, uint8_t events) {
  auto& client = clients_[index];
  assert(client.fd.get() == fd);
  if (events & Looper::EVENT_HUP) {
    close_client(index);
    return;
  }
  if (events & Looper::EVENT_ERROR) {
    logger_->out(Logger::INFO, "%zu: Client connection error", index);
    close_client(index);
    return;
  }
  if (!base_event(&client, events, index, "Client")) {
    close_client(index);
    return;
  }
  if (client.new_connection) return;
  client_empty_input(index);
}

bool setup_content(Http const* http, Content* content) {
  assert(content->type == CONTENT_NONE);
  std::string te = http->first_header("transfer-encoding");
  if (te.empty() || header_token_eq(te, "identity")) {
    std::string len = http->first_header("content-length");
    if (len.empty()) {
      content->type = CONTENT_CLOSE;
      return true;
    }
    char* end = nullptr;
    errno = 0;
    auto tmp = strtoull(len.c_str(), &end, 10);
    if (errno || !end || *end) {
      return false;
    }
    if (tmp == 0) {
      content->type = CONTENT_NONE;
      return true;
    }
    content->len = tmp;
    content->type = CONTENT_LEN;
  } else {
    content->chunked.reset(Chunked::create());
    content->type = CONTENT_CHUNKED;
  }
  return true;
}

void ProxyImpl::client_empty_input(size_t index) {
  auto& client = clients_[index];
  while (true) {
    size_t avail;
    auto ptr = client.in->read_ptr(&avail);
    if (avail == 0) return;
    switch (client.content.type) {
    case CONTENT_CLOSE:
      assert(false);
      // falltrough
    case CONTENT_NONE: {
      if (client.connect && client.remote_state == CONNECTED) {
        if (client.connect->mitm_detect == Mitm::NEED_MORE) {
          client.connect->mitm_detect = mitm_->detect(ptr, avail);
          if (client.connect->mitm_detect == Mitm::NEED_MORE) {
            // Need more data
            return;
          }
          if (client.connect->mitm_detect == Mitm::SSL) {
            client.connect->mitm.reset(mitm_->open(client.connect->host));
            if (!client.connect->mitm) {
              client.connect->mitm_detect = Mitm::OTHER;
            } else if (client.pkg_id != 0) {
              client.connect->mitm_monitor.reset(
                  new MitmMonitor(this, client.pkg_id, client.remote.pkg_id));
            }
          }
        }
        if (client.connect->mitm) {
          bool local_empty = client.out->empty();
          bool remote_empty = client.remote.out->empty();
          // TODO(the_jk): Monitor
          if (!client.connect->mitm->transfer(
                  client.in.get(), client.out.get(),
                  client.remote.in.get(), client.remote.out.get(),
                  client.connect->mitm_monitor.get())) {
            close_client(index);
            return;
          }
          if (!base_continue_send(&client, local_empty, index, "Client")) {
            close_client(index);
            return;
          }
          if (!base_continue_send(&client.remote, remote_empty, index,
                                  "Client remote")) {
            client_remote_error(index, 502);
            return;
          }
          break;
        } else {
          if (client.pkg_id != 0) {
            send_attached_data(client.pkg_id, ptr, avail, false);
          }
          if (!base_send(&client.remote, ptr, avail, index, "Client remote")) {
            client_remote_error(index, 502);
            return;
          }
          client.in->consume(avail);
          break;
        }
      }
      if (client.remote_state != CLOSED && client.remote_state != WAITING) {
        // Still working on the last request, wait
        return;
      }
      if (client.pkg_id != 0) {
        send_attached_data(client.pkg_id, nullptr, 0, true);
        client.pkg_id = 0;
      }
      client.request.reset(
          HttpRequest::parse(
              reinterpret_cast<char const*>(ptr), avail, false));
      if (!client.request) {
        if (avail >= 1024 * 1024) {
          logger_->out(Logger::INFO, "%zu: Client too large request %zu",
                       index, avail);
          close_client(index);
        }
        return;
      }
      if (!client.request->good()) {
        client_error(index, 400, "Bad request");
        return;
      }
      if (client.request->method_equal("CONNECT")) {
        client.connect.reset(new Connect());
        if (!Url::parse_authority(client.request->url(), &client.connect->host,
                                  &client.connect->port)) {
          client_error(index, 400, "Bad request");
          return;
        }
        if (mitm_) {
          client.connect->mitm_detect = Mitm::NEED_MORE;
        } else {
          client.connect->mitm_detect = Mitm::OTHER;
        }
      } else {
        client.url.reset(Url::parse(client.request->url()));
        if (!client.url) {
          client_error(index, 400, "Bad request");
          return;
        }
        if (!setup_content(client.request.get(), &client.content)) {
          logger_->out(Logger::INFO, "%zu: Client bad content-length", index);
          client_error(index, 400, "Bad request");
          return;
        }
        if (client.content.type == CONTENT_CLOSE) {
          client.content.type = CONTENT_NONE;
        }
      }
      if (!client_request(index)) {
        client.content.type = CONTENT_NONE;
        return;
      }
      break;
    }
    case CONTENT_LEN:
      if (client.remote_state < CONNECTED) {
        // Request hasn't been sent yet, still collecting data
        return;
      }
      if (avail < client.content.len) {
        if (client.pkg_id != 0) {
          send_attached_data(client.pkg_id, ptr, avail, false);
        }
        if (!base_send(&client.remote, ptr, avail, index, "Client remote")) {
          return;
        }
        client.in->consume(avail);
        client.content.len -= avail;
        return;
      }
      if (client.pkg_id != 0) {
        send_attached_data(client.pkg_id, ptr, client.content.len, true);
        client.pkg_id = 0;
      }
      if (!base_send(&client.remote, ptr, client.content.len, index,
                     "Client remote")) {
        return;
      }
      client.in->consume(client.content.len);
      client.content.len = 0;
      client.content.type = CONTENT_NONE;
      break;
    case CONTENT_CHUNKED:
      if (client.remote_state < CONNECTED) {
        // Request hasn't been sent yet, still collecting data
        return;
      }
      auto used = client.content.chunked->add(ptr, avail);
      if (!client.content.chunked->good()) {
        client_error(index, 400, "Bad request");
        return;
      }
      if (client.pkg_id != 0) {
        send_attached_data(client.pkg_id, ptr, used, false);
      }
      if (!base_send(&client.remote, ptr, used, index, "Client remote")) {
        return;
      }
      client.in->consume(used);
      if (client.content.chunked->eof()) {
        if (client.pkg_id != 0) {
          send_attached_data(client.pkg_id, nullptr, 0, true);
        }
        client.content.chunked.reset();
        client.content.type = CONTENT_NONE;
      }
      break;
    }
  }
}

void ProxyImpl::close_client_when_done(size_t index) {
  auto& client = clients_[index];
  if (client.out->empty()) {
    close_client(index);
    return;
  }
  client.remote_state = CLOSED;
  close_base(&client.remote);
  client.in.reset();
  client.read_flag = 0;
  client.request.reset();
  client.content.type = CONTENT_NONE;
  if (client.pkg_id != 0) {
    send_attached_data(client.pkg_id, nullptr, 0, true);
    client.pkg_id = 0;
  }
  looper_->modify(client.fd.get(), client.write_flag);
}

void ProxyImpl::client_remote_event(size_t index, int fd, uint8_t events) {
  auto& client = clients_[index];
  assert(client.remote.fd.get() == fd);
  if (events & Looper::EVENT_HUP) {
    logger_->out(Logger::INFO, "%zu: Client remote connection closed", index);
    client_remote_error(index, 502);
    return;
  }
  if (events & Looper::EVENT_ERROR) {
    logger_->out(Logger::INFO, "%zu: Client remote connection error", index);
    client_remote_error(index, 502);
    return;
  }
  if (client.remote_state == CONNECTING) {
    if (events & Looper::EVENT_WRITE) {
      if (client.connect) {
        auto req = std::unique_ptr<HttpResponseBuilder>(
            HttpResponseBuilder::create(
                client.request->proto(),
                client.request->proto_version(),
                200, "OK"));
        auto data = req->build();
        client.in->consume(client.request->size());
        client.request.reset();
        client.remote.content.type = CONTENT_CLOSE;
        client.remote.pkg_id = get_next_package_id();
        if (client.remote.pkg_id) {
          send_attached_package(client.remote.pkg_id, 0,
                                client.remote.host, client.remote.port,
                                client.source_host, client.source_port, false);
        }
        if (!base_send(&client, data.data(), data.size(), index, "Client")) {
          close_client(index);
          return;
        }
        events &= ~Looper::EVENT_WRITE;
      } else {
        std::string url(client.url->path_escaped());
        if (url.empty()) url.push_back('/');
        auto query = client.url->full_query_escaped();
        if (query) {
          url.push_back('?');
          url.append(query);
        }
        auto req = std::unique_ptr<HttpRequestBuilder>(
            HttpRequestBuilder::create(
                client.request->method(),
                url,
                client.request->proto(),
                client.request->proto_version()));
        auto iter = client.request->header();
        bool have_host = false;
        for (; iter->valid(); iter->next()) {
          if (!have_host && iter->name_equal("host")) have_host = true;
          if (iter->name_equal("proxy-connection") ||
              iter->name_equal("proxy-authenticate") ||
              iter->name_equal("proxy-authorization")) {
            continue;
          }
          req->add_header(iter->name(), iter->value());
        }
        if (!have_host &&
            (client.request->proto_version().major == 1 &&
             client.request->proto_version().minor == 1)) {
          req->add_header("host", client.url->host());
        }
        auto data = req->build();
        if (client.pkg_id != 0) {
          bool const last = client.content.type == CONTENT_NONE;
          if (monitor_send_proxied_) {
            send_attached_data(client.pkg_id, data.data(), data.size(), last);
          } else {
            auto ptr = client.in->read_ptr(nullptr);
            send_attached_data(client.pkg_id,
                               ptr, client.request->size(), last);
          }
          if (last) {
            client.pkg_id = 0;
          }
        }
        client.in->consume(client.request->size());
        client.request.reset();
        client.url.reset();
        client.remote.out->write(data.data(), data.size());

        client.remote.pkg_id = get_next_package_id();
        if (client.remote.pkg_id) {
          send_attached_package(client.remote.pkg_id, 0,
                                client.remote.host, client.remote.port,
                                client.source_host, client.source_port, false);
        }
      }
      client.remote_state = CONNECTED;
      client.remote.read_flag = Looper::EVENT_READ;
      client.remote.write_flag =
        client.remote.out->empty() ? 0 : Looper::EVENT_WRITE;
      looper_->modify(client.remote.fd.get(),
                      client.remote.read_flag | client.remote.write_flag);
      client_empty_input(index);
    } else {
      return;
    }
  }
  if (!base_event(&client.remote, events, index, "Client remote")) {
    switch (client.remote_state) {
    case CONNECTED:
      switch (client.remote.content.type) {
      case CONTENT_CLOSE:
        close_client_when_done(index);
        break;
      case CONTENT_CHUNKED:
      case CONTENT_LEN:
      case CONTENT_NONE:
        client_remote_error(index, 502);
        break;
      }
      break;
    case CONNECTING:
    case RESOLVING:
    case CLOSED:
      assert(false);
      client_remote_error(index, 502);
      break;
    case WAITING:
      if (client.remote.pkg_id != 0) {
        send_attached_data(client.remote.pkg_id, nullptr, 0, true);
        client.remote.pkg_id = 0;
      }
      close_base(&client.remote);
      client.remote_state = CLOSED;
      break;
    }
    return;
  }
  while (true) {
    size_t avail;
    auto ptr = client.remote.in->read_ptr(&avail);
    if (avail == 0) return;
    switch (client.remote_state) {
    case CONNECTED:
      switch (client.remote.content.type) {
      case CONTENT_NONE: {
        auto response = std::unique_ptr<HttpResponse>(
            HttpResponse::parse(
                reinterpret_cast<char const*>(ptr), avail, false));
        if (!response) {
          if (avail >= 1024 * 1024) {
            logger_->out(Logger::INFO,
                         "%zu: Client remote too large request %zu",
                         index, avail);
            client_remote_error(index, 502);
          }
          return;
        }
        if (!response->good()) {
          client_remote_error(index, 502);
          return;
        }
        if (!setup_content(response.get(), &client.remote.content)) {
          logger_->out(Logger::INFO, "%zu: Client remote bad content-length",
                       index);
          client.remote.content.type = CONTENT_CLOSE;
        } else {
          if (client.remote.content.type == CONTENT_NONE) {
            client.remote_state = WAITING;
            client.remote.read_flag = 0;
            client.remote.write_flag = 0;
            looper_->modify(client.remote.fd.get(), 0);
          }
        }
        if (client.remote.pkg_id != 0) {
          if (client.remote.content.type == CONTENT_NONE) {
            send_attached_data(client.remote.pkg_id,
                               ptr, response->size(), true);
            client.remote.pkg_id = 0;
          } else {
            send_attached_data(client.remote.pkg_id,
                               ptr, response->size(), false);
          }
        }
        if (!base_send(&client, ptr, response->size(), index, "Client")) {
          close_client(index);
          return;
        }
        client.remote.in->consume(response->size());
        break;
      }
      case CONTENT_LEN:
        if (avail < client.remote.content.len) {
          if (client.remote.pkg_id != 0) {
            send_attached_data(client.remote.pkg_id, ptr, avail, false);
          }
          if (!base_send(&client, ptr, avail, index, "Client")) {
            close_client(index);
            return;
          }
          client.remote.in->consume(avail);
          client.remote.content.len -= avail;
          return;
        }
        if (client.remote.pkg_id != 0) {
          send_attached_data(client.remote.pkg_id, ptr,
                             client.remote.content.len, true);
          client.remote.pkg_id = 0;
        }
        if (!base_send(&client, ptr, client.remote.content.len,
                       index, "Client")) {
          close_client(index);
          return;
        }
        client.remote.in->consume(client.remote.content.len);
        client.remote.content.len = 0;
        client.remote.content.type = CONTENT_NONE;
        client.remote_state = WAITING;
        client.remote.read_flag = 0;
        client.remote.write_flag = 0;
        looper_->modify(client.remote.fd.get(), 0);
        return;
      case CONTENT_CHUNKED: {
        auto used = client.remote.content.chunked->add(ptr, avail);
        if (!client.remote.content.chunked->good()) {
          logger_->out(Logger::INFO, "%zu: Client remote bad chunked",
                       index);
          client.remote.content.type = CONTENT_CLOSE;
          break;
        }
        if (client.remote.pkg_id != 0) {
          send_attached_data(client.remote.pkg_id, ptr, used, false);
        }
        if (!base_send(&client, ptr, used, index, "Client")) {
          close_client(index);
          return;
        }
        client.remote.in->consume(used);
        if (client.remote.content.chunked->eof()) {
          if (client.remote.pkg_id != 0) {
            send_attached_data(client.remote.pkg_id, nullptr, 0, true);
            client.remote.pkg_id = 0;
          }
          client.remote.content.chunked.reset();
          client.remote.content.type = CONTENT_NONE;
          client.remote_state = WAITING;
          client.remote.read_flag = 0;
          client.remote.write_flag = 0;
          looper_->modify(client.remote.fd.get(), 0);
        }
        break;
      }
      case CONTENT_CLOSE:
        if (client.connect) {
          if (client.connect->mitm_detect == Mitm::NEED_MORE) {
            // Need more data (on client side)
            return;
          }
          if (client.connect->mitm) {
            bool local_empty = client.out->empty();
            bool remote_empty = client.remote.out->empty();
            // TODO(the_jk): Monitor
            if (!client.connect->mitm->transfer(
                    client.in.get(), client.out.get(),
                    client.remote.in.get(), client.remote.out.get(),
                    client.connect->mitm_monitor.get())) {
              close_client(index);
              return;
            }
            if (!base_continue_send(&client, local_empty, index, "Client")) {
              return;
            }
            if (!base_continue_send(&client.remote, remote_empty, index,
                                    "Client remote")) {
              client_remote_error(index, 502);
              return;
            }
            break;
          }
        }
        if (client.remote.pkg_id != 0) {
          send_attached_data(client.remote.pkg_id, ptr, avail, false);
        }
        if (!base_send(&client, ptr, avail, index, "Client")) {
          close_client(index);
          return;
        }
        client.remote.in->consume(avail);
        break;
      }
      break;
    case CONNECTING:
    case RESOLVING:
    case CLOSED:
      assert(false);
      return;
    case WAITING:
      return;
    }
  }
}

void ProxyImpl::monitor_error(size_t index, uint16_t status_code,
                              std::string const& status_message) {
  auto& monitor = monitors_[index];
  // No more input
  monitor.read_flag = 0;
  looper_->modify(monitor.fd.get(), monitor.write_flag);

  auto resp = std::unique_ptr<HttpResponseBuilder>(
      HttpResponseBuilder::create(
          "HTTP", Version(1, 0), status_code, status_message));
  resp->add_header("Content-Length", "0");
  resp->add_header("Connection", "close");

  monitor.in.reset();

  auto data = resp->build();

  if (!base_send(&monitor, data.data(), data.size(), index, "Monitor")) {
    close_monitor(index);
  }
}

bool ProxyImpl::support_monitor_version(
    size_t UNUSED(index), std::string const& version) {
  if (version.empty()) return false;
  // TODO: Actually do some version check here
  return version.compare(VERSION) == 0;
}

bool ProxyImpl::monitor_send_chunked(
    size_t index, void const* header, size_t header_size,
    void const* data, size_t data_size) {
  auto& monitor = monitors_[index];
  char chunked_header[10];
  auto len = snprintf(chunked_header, 10, "%zx\r\n", header_size + data_size);
  if (!base_send(&monitor, chunked_header, len, index, "Monitor")) return false;
  if (header_size > 0 && !base_send(&monitor,
                                    header, header_size, index, "Monitor")) {
    return false;
  }
  if (data_size > 0
      && !base_send(&monitor, data, data_size, index, "Monitor")) {
    return false;
  }
  return base_send(&monitor, "\r\n", 2, index, "Monitor");
}

void ProxyImpl::monitor_event(size_t index, int fd, uint8_t events) {
  auto& monitor = monitors_[index];
  assert(monitor.fd.get() == fd);
  if (events & Looper::EVENT_HUP) {
    close_monitor(index);
    return;
  }
  if (events & Looper::EVENT_ERROR) {
    logger_->out(Logger::INFO, "%zu: Monitor connection error", index);
    close_monitor(index);
    return;
  }
  if (!base_event(&monitor, events, index, "Monitor")) {
    close_monitor(index);
    return;
  }

  while (true) {
    size_t avail;
    auto ptr = monitor.in->read_ptr(&avail);
    if (avail == 0) return;
    auto request = std::unique_ptr<HttpRequest>(
        HttpRequest::parse(
            reinterpret_cast<char const*>(ptr), avail, false));
    if (!request) {
      if (avail >= 1024 * 1024) {
        logger_->out(Logger::INFO, "%zu: Monitor too large request %zu",
                     index, avail);
        close_monitor(index);
      }
      return;
    }
    if (!request->good()
        || !request->proto_equal("HTTP")
        || !request->method_equal("GET")
        // Only support 1.1 or above
        || request->proto_version().major == 0
        || (request->proto_version().major == 1
            && request->proto_version().minor == 0)) {
      monitor_error(index, 400, "Bad request");
      return;
    }
    Content content;
    content.type = CONTENT_NONE;
    if (!setup_content(request.get(), &content)) {
      monitor_error(index, 400, "Bad request");
      return;
    }
    if (content.type != CONTENT_NONE) {
      monitor_error(index, 400, "Bad request");
      return;
    }
    std::unique_ptr<HttpResponseBuilder> resp;
    if (request->url_equal("/hello")) {
      auto version = request->first_header("x-tp-monitor-version");
      if (support_monitor_version(index, version)) {
        resp.reset(HttpResponseBuilder::create(
                       "HTTP", Version(1, 1), 200, "OK"));
        monitor.got_hello = true;
      } else {
        resp.reset(HttpResponseBuilder::create(
                       "HTTP", Version(1, 1), 500, "Unsupported version"));
      }
      resp->add_header("X-TP-Version", VERSION);
      resp->add_header("Content-Length", "0");
    } else if (!monitor.got_hello) {
      monitor_error(index, 500, "Unexpected request");
      return;
    } else if (request->url_equal("/attach")) {
      attached_.insert(index);
      resp.reset(HttpResponseBuilder::create(
                     "HTTP", Version(1, 1), 200, "OK"));
      resp->add_header("Transfer-Encoding", "chunked");
    } else if (request->url_equal("/detach")) {
      auto it = attached_.find(index);
      if (it != attached_.end()) {
        if (!monitor_send_chunked(index, nullptr, 0, nullptr, 0)) {
          close_monitor(index);
          return;
        }
        attached_.erase(it);
      }
      resp.reset(HttpResponseBuilder::create(
                     "HTTP", Version(1, 1), 200, "OK"));
      resp->add_header("Content-Length", "0");
    } else {
      resp.reset(HttpResponseBuilder::create(
                     "HTTP", Version(1, 1), 404, "Not found"));
      resp->add_header("Content-Length", "0");
    }
    auto data = resp->build();
    monitor.in->consume(request->size());
    if (!base_send(&monitor, data.data(), data.size(), index, "Monitor")) {
      close_monitor(index);
      return;
    }
  }
}

void ProxyImpl::new_base(BaseClient* client, int fd) {
  client->fd.reset(fd);
  client->new_connection = true;
  client->read_flag = Looper::EVENT_READ;
  client->write_flag = 0;
  client->last = looper_->now();
  if (!new_timeout_) {
    new_timeout_ = looper_->schedule(
        NEW_CONNECTION_TIMEOUT.count(),
        std::bind(&ProxyImpl::new_timeout, this));
  }
}

#ifndef HAVE_ACCEPT4
# define accept4 my_accept4
# ifdef SOCK_NONBLOCK
#  undef SOCK_NONBLOCK
# endif
# define SOCK_NONBLOCK 1
int my_accept4(int sockfd, struct sockaddr* addr, socklen_t* addrlen,
               int flags) {
  auto ret = accept(sockfd, addr, addrlen);
  if (ret != -1 && (flags & SOCK_NONBLOCK)) {
    if (fcntl(ret, F_SETFL, O_NONBLOCK)) {
      close(ret);
      return -1;
    }
  }
  return ret;
}
#endif  // HAVE_ACCEPT4

union big_addr {
  struct sockaddr_in addr_in;
  struct sockaddr_in6 addr_in6;
};

void ProxyImpl::new_client(int fd, uint8_t events) {
  assert(fd == accept_socket_.get());
  if (events == Looper::EVENT_READ) {
    assert(!clients_.full());
    while (true) {
      big_addr addr;
      socklen_t len = sizeof(addr);
      auto a = reinterpret_cast<struct sockaddr*>(&addr);
      int ret = accept4(fd, a, &len, SOCK_NONBLOCK);
      if (ret < 0) {
        if (errno == EAGAIN || errno == EWOULDBLOCK) return;
        if (errno == EINTR) continue;
        logger_->out(Logger::WARN, "Accept failed: %s", strerror(errno));
        return;
      }
      auto index = clients_.new_client();
      new_base(&clients_[index], ret);
      clients_[index].source_host.clear();
      if (len == sizeof(struct sockaddr_in) && a->sa_family == AF_INET) {
        char tmp[INET_ADDRSTRLEN];
        if (inet_ntop(AF_INET, &addr.addr_in.sin_addr, tmp, sizeof(tmp))) {
          clients_[index].source_host = tmp;
        }
        clients_[index].source_port = ntohs(addr.addr_in.sin_port);
      } else if (len == sizeof(struct sockaddr_in6)
               && a->sa_family == AF_INET6) {
        char tmp[INET6_ADDRSTRLEN];
        if (inet_ntop(AF_INET6, &addr.addr_in6.sin6_addr, tmp, sizeof(tmp))) {
          clients_[index].source_host = tmp;
        }
        clients_[index].source_port = ntohs(addr.addr_in6.sin6_port);
      } else {
        clients_[index].source_port = 0;
      }
      clients_[index].content.type = CONTENT_NONE;
      clients_[index].remote_state = CLOSED;
      clients_[index].remote.content.type = CONTENT_NONE;
      clients_[index].pkg_id = 0;
      clients_[index].remote.pkg_id = 0;
      looper_->add(ret, clients_[index].read_flag | clients_[index].write_flag,
                   std::bind(&ProxyImpl::client_event, this,
                             index,
                             std::placeholders::_1,
                             std::placeholders::_2));
      break;
    }
    if (clients_.full()) looper_->modify(fd, 0);
  } else {
    logger_->out(Logger::ERR, "Accept socket died");
    fatal_error();
  }
}

void ProxyImpl::setup_monitor(int fd) {
  auto index = monitors_.new_client();
  new_base(&monitors_[index], fd);
  monitors_[index].got_hello = false;
  looper_->add(fd, monitors_[index].read_flag
               | monitors_[index].write_flag,
               std::bind(&ProxyImpl::monitor_event, this,
                         index,
                         std::placeholders::_1,
                         std::placeholders::_2));
}

void ProxyImpl::new_monitor(int fd, uint8_t events) {
  assert(fd == monitor_socket_.get());
  if (events == Looper::EVENT_READ) {
    assert(!monitors_.full());
    while (true) {
      int ret = accept4(fd, nullptr, nullptr, SOCK_NONBLOCK);
      if (ret < 0) {
        if (errno == EAGAIN || errno == EWOULDBLOCK) return;
        if (errno == EINTR) continue;
        logger_->out(Logger::WARN, "Accept failed: %s", strerror(errno));
        return;
      }
      setup_monitor(ret);
      break;
    }
    if (monitors_.full()) looper_->modify(fd, 0);
  } else {
    logger_->out(Logger::ERR, "Monitor socket died");
    fatal_error();
  }
}

void ProxyImpl::fatal_error() {
  looper_->quit();
  good_ = false;
}

bool ProxyImpl::run() {
  good_ = true;
  if (!logger_) {
    priv_logger_.reset(Logger::create_syslog("tp"));
    logger_ = priv_logger_.get();
  }
  {
    struct sigaction action;
    memset(&action, 0, sizeof(action));
    action.sa_handler = SIG_IGN;
    action.sa_flags = SA_RESTART;
    sigaction(SIGPIPE, &action, nullptr);
  }
  if (!signal_pipe_.open()) {
    logger_->out(Logger::WARN,
                 "Failed to create pipes, signals wont work: %s",
                 strerror(errno));
  }
  if (signal_pipe_) {
    looper_->add(signal_pipe_.read(), Looper::EVENT_READ,
                 std::bind(&ProxyImpl::signal_event, this,
                           std::placeholders::_1,
                           std::placeholders::_2));
  }
  if (!looper_->run()) {
    logger_->out(Logger::ERR, "poll() failed: %s", strerror(errno));
    return false;
  }
  return good_;
}

uint32_t ProxyImpl::get_next_package_id() {
  if (attached_.empty()) return 0;
  while (!next_package_id_) {
    ++next_package_id_;
  }
  return next_package_id_++;
}

void ProxyImpl::send_attached_package(uint32_t id, uint16_t flags,
                                      std::string const& source_host,
                                      uint16_t source_port,
                                      std::string const& target_host,
                                      uint16_t target_port,
                                      bool last) {
  if (id == 0) {
    assert(false);
    return;
  }
  if (attached_.empty()) return;
  uint8_t data[256];

  Package pkg;
  pkg.id = id;
  auto dur = looper_->now().time_since_epoch();
  auto sec = std::chrono::duration_cast<std::chrono::seconds>(dur);
  pkg.timestamp.tv_sec = sec.count();
  pkg.timestamp.tv_nsec =
    std::chrono::duration_cast<std::chrono::nanoseconds>(dur - sec).count();
  pkg.flags = (last ? 0 : 1) | (flags << 1);
  pkg.source_port = source_port;
  pkg.source_host = source_host;
  pkg.target_port = target_port;
  pkg.target_host = target_host;

  auto size = 5 + write_package(pkg, data + 5, sizeof(data) - 5);
  auto ptr = data;
  std::unique_ptr<uint8_t[]> extra;
  if (size > sizeof(data)) {
    // TODO: Might need better handling of really long source_host/target_host
    extra.reset(new uint8_t[size]);
    ptr = extra.get();
    write_package(pkg, ptr + 5, size - 5);
  }
  ptr[0] = 'P';
  ptr[1] = 'K';
  ptr[2] = 'G';
  write_u16(ptr + 3, size);
  send_attached(ptr, size, nullptr, 0);
}

void ProxyImpl::send_attached_data(uint32_t id, void const* ptr, size_t size,
                                   bool last) {
  if (id == 0) {
    assert(false);
    return;
  }
  if (attached_.empty()) return;
  if (size == 0 && !last) return;
  uint8_t data[10];
  data[0] = 'D';
  data[1] = 'A';
  data[2] = 'T';
  write_u32(data + 5, id);
  if (size == 0) {
    assert(last);
    assert(ptr == nullptr);
    write_u16(data + 3, 10);
    data[9] = 0;
    send_attached(data, 10, nullptr, 0);
  } else {
    size_t max = 0xffff - 10;
    auto p = reinterpret_cast<char const*>(ptr);
    data[9] = 1;
    while (size > max) {
      write_u16(data + 3, 0xffff);
      send_attached(data, 10, p, max);
      p += max;
      size -= max;
    }
    data[9] = last ? 0 : 1;
    write_u16(data + 3, size + 10);
    send_attached(data, 10, p, size);
  }
}

void ProxyImpl::send_attached(void const* header, size_t header_size,
                              void const* data, size_t data_size) {
  auto it = attached_.begin();
  while (it != attached_.end()) {
    if (monitor_send_chunked(*it, header, header_size, data, data_size)) {
      ++it;
    } else {
      auto index = *it;
      it = attached_.erase(it);
      close_monitor(index);
    }
  }
}

int setup_socket(char const* host, std::string const& port, Logger* logger) {
  io::auto_fd ret;
  struct addrinfo hints;
  memset(&hints, 0, sizeof(hints));
  hints.ai_family = AF_UNSPEC;
  hints.ai_socktype = SOCK_STREAM;
  hints.ai_flags = AI_V4MAPPED | AI_ADDRCONFIG | AI_PASSIVE;
  struct addrinfo* result;
  auto retval = getaddrinfo(host && *host ? host : nullptr, port.c_str(),
                            &hints, &result);
  if (retval) {
    logger->out(Logger::ERR, "getaddrinfo failed for %s:%s: %s",
                host ? host : "*", port.c_str(), gai_strerror(retval));
    return -1;
  }
  auto retp = result;
  for (; retp; retp = retp->ai_next) {
    ret.reset(socket(retp->ai_family, retp->ai_socktype,
                               retp->ai_protocol));
    if (!ret) continue;
    if (bind(ret.get(), retp->ai_addr, retp->ai_addrlen) == 0)
      break;
  }
  freeaddrinfo(result);
  if (!retp) {
    logger->out(Logger::ERR, "Failed to bind %s:%s: %s",
                host ? host : "*", port.c_str(), strerror(errno));
    return -1;
  }
  if (listen(ret.get(), SOMAXCONN)) {
    logger->out(Logger::ERR, "Failed to listen: %s", strerror(errno));
    return -1;
  }
  if (fcntl(ret.get(), F_SETFL, O_NONBLOCK)) {
    logger->out(Logger::ERR, "fcntl(O_NONBLOCK) failed: %s", strerror(errno));
    return -1;
  }
  return ret.release();
}

}  // namespace

// static
Proxy* Proxy::create(Config* config, std::string const& cwd,
                     char const* configfile,
                     char const* logfile,
                     Logger* logger,
                     int accept_fd,
                     int monitor_fd,
                     Looper* looper, Resolver* resolver) {
  return new ProxyImpl(config, cwd, configfile, logfile, logger,
                       accept_fd, monitor_fd, looper, resolver);
}

// static
int Proxy::setup_accept_socket(Config* config, Logger* logger) {
  return setup_socket(config->get("proxy_bind", nullptr),
                      config->get("proxy_port", "8080"), logger);
}

// static
int Proxy::setup_monitor_socket(Config* config, Logger* logger) {
  assert(config->get("monitor", false));
  return setup_socket(config->get("monitor_bind", "localhost"),
                      config->get("monitor_port", "9000"), logger);
}