#include "http.h"

#include <boost/beast/version.hpp>

// Support both boost in Debian unstable (BOOST_LATEST) and in stable (boost 1.67)
#if BOOST_VERSION >= 107100
#define BOOST_LATEST
#endif

#include "server.h"
#include "response.h"

#include <boost/beast/version.hpp>
#include <boost/beast/core.hpp>
#include <boost/beast/http.hpp>
#include <boost/asio/dispatch.hpp>
#ifndef BOOST_LATEST
#include <boost/asio/bind_executor.hpp>
#include <boost/asio/ip/tcp.hpp>
#endif
#include <boost/asio/strand.hpp>
#include <boost/config.hpp>
#include <algorithm>
#include <cstdlib>
#include <functional>
#include <iostream>
#include <memory>
#include <optional>
#include <string>
#include <thread>
#include <vector>

namespace beast = boost::beast;         // from <boost/beast.hpp>
namespace http = beast::http;           // from <boost/beast/http.hpp>
namespace net = boost::asio;            // from <boost/asio.hpp>
using tcp = boost::asio::ip::tcp;       // from <boost/asio/ip/tcp.hpp>

namespace {

//------------------------------------------------------------------------------

// Report a failure
void fail(beast::error_code ec, char const* what)
{
 std::cerr << what << ": " << ec.message() << "\n";
}

// Handles an HTTP server connection
class session : public std::enable_shared_from_this<session>
{
#ifdef BOOST_LATEST
 beast::tcp_stream stream_;
#else
 tcp::socket socket_;
 boost::asio::strand<boost::asio::io_context::executor_type> strand_;
#endif
 beast::flat_buffer buffer_;
 Server& m_server;
 std::optional<http::request_parser<http::string_body>> parser_;
 request_type req_;
 std::shared_ptr<response_type> res_; // std::shared_ptr<void> ?

 void handle_request(::Server& server, request_type&& req)
 {
#ifdef BOOST_LATEST
  stream_.expires_after(std::chrono::seconds(300)); // timeout on write by server much longer than read timeout from client
#else
  // socket_.expires_after(std::chrono::seconds(300)); // not supported by old boost
#endif
  auto sp = std::make_shared<response_type>(generate_response(req, server));

  res_ = sp;

  // Write the response
#ifdef BOOST_LATEST
  http::async_write(
      stream_,
      *sp,
      beast::bind_front_handler(
          &session::on_write,
          shared_from_this(),
          sp->need_eof()));
#else
  http::async_write(
      socket_,
      *sp,
      boost::asio::bind_executor(
	  strand_,
	  std::bind(
	      &session::on_write,
	      shared_from_this(),
	      std::placeholders::_1,
	      std::placeholders::_2,
	      sp->need_eof())));
#endif
 }

public:
    // Take ownership of the stream
    session(
#ifdef BOOST_LATEST
        tcp::socket&& socket,
#else
        tcp::socket socket,
#endif
        Server& server)
#ifdef BOOST_LATEST
        : stream_(std::move(socket))
#else
        : socket_(std::move(socket))
        , strand_(socket_.get_executor())
#endif
        , m_server(server)
    {
    }

    // Start the asynchronous operation
    void run()
    {
        // We need to be executing within a strand to perform async operations
        // on the I/O objects in this session.
#ifdef BOOST_LATEST
        net::dispatch(stream_.get_executor(),
                      beast::bind_front_handler(
                          &session::do_read,
                          shared_from_this()));
#else
        do_read();
#endif
    }

    void do_read()
    {
        // Make the request empty before reading,
        // otherwise the operation behavior is undefined.
        req_ = {};
       
        // this is the way to reset the parser. it's necessary.
        // https://github.com/boostorg/beast/issues/927
        parser_.emplace();
        parser_->body_limit(1000000000); // 1GB limit

#ifdef BOOST_LATEST
        // Set the timeout.
        stream_.expires_after(std::chrono::seconds(30));

        // Read a request
        http::async_read(stream_, buffer_, *parser_,
            beast::bind_front_handler(
                &session::on_read,
                shared_from_this()));
#else

        http::async_read(socket_, buffer_, *parser_,
            boost::asio::bind_executor(
                strand_,
                std::bind(
                    &session::on_read,
                    shared_from_this(),
                    std::placeholders::_1,
                    std::placeholders::_2)));
#endif
    }

    void
    on_read(
#ifdef BOOST_LATEST
        beast::error_code ec,
        std::size_t bytes_transferred
#else
        boost::system::error_code ec,
        std::size_t bytes_transferred
#endif
        )
    {
        boost::ignore_unused(bytes_transferred);

        // This means they closed the connection
        if (ec == http::error::end_of_stream)
            return do_close();

        if (ec == http::error::partial_message)
         return; // ignore

        if (ec)
            return fail(ec, "read");

        req_ = parser_->get();
        
        // Send the response
        handle_request(m_server, std::move(req_));
    }

    void
    on_write(
#ifdef BOOST_LATEST
        bool close,
        beast::error_code ec,
        std::size_t bytes_transferred
#else
        boost::system::error_code ec,
        std::size_t bytes_transferred,
        bool close
#endif
        )
    {
        boost::ignore_unused(bytes_transferred);

        if(ec)
            return fail(ec, "write");

        if(close)
        {
            // This means we should close the connection, usually because
            // the response indicated the "Connection: close" semantic.
            return do_close();
        }

        // We're done with the response so delete it
        res_ = nullptr;

        // Read another request
        do_read();
    }

    void
    do_close()
    {
        // Send a TCP shutdown
        beast::error_code ec;
#ifdef BOOST_LATEST
        stream_.socket().shutdown(tcp::socket::shutdown_send, ec);
#else
        socket_.shutdown(tcp::socket::shutdown_send, ec);
#endif
        // At this point the connection is closed gracefully
    }
};

//------------------------------------------------------------------------------

// Accepts incoming connections and launches the sessions
class listener : public std::enable_shared_from_this<listener>
{
#ifdef BOOST_LATEST
    net::io_context& ioc_;
#endif
    tcp::acceptor acceptor_;
#ifndef BOOST_LATEST
    tcp::socket socket_;
#endif
    Server& m_server;

public:
    listener(
        net::io_context& ioc,
        tcp::endpoint endpoint,
        Server& server)
#ifdef BOOST_LATEST
        : ioc_(ioc)
        , acceptor_(net::make_strand(ioc))
#else
        : acceptor_(ioc)
        , socket_(ioc)
#endif
        , m_server(server)
    {
#ifdef BOOST_VERSION
        beast::error_code ec;
#else
        boost::system::error_code ec;
#endif

        // Open the acceptor
        acceptor_.open(endpoint.protocol(), ec);
        if(ec)
        {
            fail(ec, "open");
            return;
        }

        // Allow address reuse
        acceptor_.set_option(net::socket_base::reuse_address(true), ec);
        if(ec)
        {
            fail(ec, "set_option");
            return;
        }

        // Bind to the server address
        acceptor_.bind(endpoint, ec);
        if(ec)
        {
            fail(ec, "bind");
            return;
        }

        // Start listening for connections
        acceptor_.listen(
            net::socket_base::max_listen_connections, ec);
        if(ec)
        {
            fail(ec, "listen");
            return;
        }
    }

    // Start accepting incoming connections
    void
    run()
    {
#ifndef BOOST_LATEST
     if (!acceptor_.is_open())
       return;
#endif
     do_accept();
    }

private:
    void
    do_accept()
    {
        // The new connection gets its own strand
#ifdef BOOST_LATEST
        acceptor_.async_accept(
            net::make_strand(ioc_),
            beast::bind_front_handler(
                &listener::on_accept,
                shared_from_this()));
#else
        acceptor_.async_accept(
            socket_,
            std::bind(
                &listener::on_accept,
                shared_from_this(),
                std::placeholders::_1));
#endif
    }

    void
#ifdef BOOST_LATEST
    on_accept(beast::error_code ec, tcp::socket socket)
#else
    on_accept(boost::system::error_code ec)
#endif
    {
        if(ec)
        {
            fail(ec, "accept");
        }
        else
        {
            // Create the session and run it
            std::make_shared<session>(
#ifdef BOOST_LATEST
                std::move(socket),
#else
                std::move(socket_),
#endif
                m_server)->run();
        }

        // Accept another connection
        do_accept();
    }
};

} // anonymous namespace

//------------------------------------------------------------------------------

namespace HTTP {

 Server::Server(Config& config, boost::asio::io_context& ioc, const Socket& socket, plugins_container_type& plugins, Statistics& statistics)
  : ::Server(config, ioc, socket, plugins, statistics)
 {
 }

 Server::~Server()
 {
 }

 int Server::start()
 {
  auto const address = net::ip::make_address(m_socket.address);
  auto const port = static_cast<unsigned short>(std::atoi(m_socket.port.data()));

  // Create and launch a listening port
  std::make_shared<listener>(
   m_ioc,
   tcp::endpoint{address, port},
   *this)->run();

  return EXIT_SUCCESS;
 }

} // namespace HTTP