path: root/server/src/srv.cpp

module;

#include <boost/asio/as_tuple.hpp>
#include <boost/asio/awaitable.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/beast/core.hpp>
#include <boost/beast/http.hpp>
#include <boost/beast/zlib.hpp>
#include <boost/config.hpp>
#include <boost/json.hpp>
#include <boost/locale/generator.hpp>

#include <expat.h>

module routemon:srv$impl;

import std;
import :gpx;
import :srv;

namespace beast = boost::beast;
namespace chrono = std::chrono;
namespace json = boost::json;
namespace net = boost::asio;
using tcp = boost::asio::ip::tcp;

namespace routemon::srv {

class gpx_parse_error_category_impl : public std::error_category
{
public:
  char const* name() const noexcept override { return "gpx_parse"; }

  auto message(int condition) const noexcept -> std::string override
  {
    std::ignore = condition;
    return "failed to parse GPX file";
  }
};

auto gpx_parse_error_category() noexcept -> gpx_parse_error_category_impl const&
{
  static auto const inst = gpx_parse_error_category_impl{};
  return inst;
}

auto gpx_parse_error() noexcept -> std::error_code
{
  return std::error_code{1, gpx_parse_error_category()};
}

class gpx_parse_result
{
  chrono::steady_clock::duration init_parse_dur_;
  chrono::steady_clock::duration put_parse_dur_;
  chrono::steady_clock::duration finish_parse_dur_;
  std::variant<std::exception_ptr, gpx::file> res_;

public:
  auto set_exception(std::exception_ptr ex) noexcept { res_ = ex; }
  auto set_gpx_file(gpx::file&& f) noexcept { res_ = std::move(f); }

  auto init_parse_duration(chrono::steady_clock::duration dur) noexcept
  {
    init_parse_dur_ = dur;
  }
  [[nodiscard]] auto init_parse_duration() const noexcept
      -> chrono::steady_clock::duration
  {
    return init_parse_dur_;
  }
  auto put_parse_duration(chrono::steady_clock::duration dur) noexcept
  {
    put_parse_dur_ = dur;
  }
  [[nodiscard]] auto put_parse_duration() const noexcept
      -> chrono::steady_clock::duration
  {
    return put_parse_dur_;
  }
  auto finish_parse_duration(chrono::steady_clock::duration dur) noexcept
  {
    finish_parse_dur_ = dur;
  }
  [[nodiscard]] auto finish_parse_duration() const noexcept
      -> chrono::steady_clock::duration
  {
    return finish_parse_dur_;
  }

  auto unwrap() -> gpx::file&&
  {
    return std::visit(
        util::overloaded{
          [](std::exception_ptr ex) -> gpx::file&&
          {
            if (ex)
              std::rethrow_exception(ex);
            else
              throw std::runtime_error{"no GPX file parse result available"};
          },
          [](gpx::file&& f) -> gpx::file&& { return std::move(f); },
        },
        std::move(res_));
  }
};

struct readable_gpx_body
{
  using value_type = gpx_parse_result;

  class reader
  {
    gpx::reader r_;
    util::not_null<value_type*> res_;
    std::inplace_vector<char, 4096> buf_;

  public:
    template <bool isRequest, bhttp::concepts::fields Fields>
    explicit reader(bhttp::header<isRequest, Fields>&, value_type& v) : res_{&v}
    {
      res_->init_parse_duration({});
      res_->put_parse_duration({});
      res_->finish_parse_duration({});
    }

    // The following methods (which are called by Beast) are marked
    // noexcept, since Beast does not ensure that exceptions thrown
    // here are appropriately directed to the caller of
    // (async_)read(_some), so throwing here might cause the program
    // to crash.

    auto
    init(boost::optional<std::uint64_t> /* n */, beast::error_code& ec) noexcept
        -> void
    {
      auto const init_start = chrono::steady_clock::now();
      try
      {
        r_.init();
        ec = {};
      }
      catch (std::exception& ex)
      {
        res_->set_exception(std::current_exception());
        ec = gpx_parse_error();
      }
      res_->init_parse_duration(chrono::steady_clock::now() - init_start);
    }

    auto
    put(beast::concepts::const_buffer_sequence auto b,
        beast::error_code& ec) noexcept -> std::size_t
    {
      auto const put_start = chrono::steady_clock::now();
      auto total = 0uz;
      try
      {
        for (auto cur_in_buf : beast::buffers_range(b))
        {
          while (cur_in_buf.size() > 0)
          {
            auto to_read =
                std::min(cur_in_buf.size(), buf_.max_size() - buf_.size());
            buf_.append_range(
                std::span{
                  static_cast<char const*>(cur_in_buf.data()), to_read
                });
            cur_in_buf += to_read;
            total += to_read;

            if (buf_.size() == buf_.max_size())
            {
              r_.put(std::string_view{buf_});
              buf_.clear();
            }
          }
        }
        ec = {};
      }
      catch (std::exception& ex)
      {
        res_->set_exception(std::current_exception());
        ec = gpx_parse_error();
      }
      res_->put_parse_duration(
          res_->put_parse_duration()
          + (chrono::steady_clock::now() - put_start));
      return total;
    }

    auto finish(beast::error_code& ec) noexcept
    {
      auto const finish_start = chrono::steady_clock::now();
      try
      {
        if (buf_.size() > 0)
        {
          r_.put(std::string_view{buf_});
          buf_.clear();
        }

        res_->set_gpx_file(r_.finish());
        ec = {};
      }
      catch (std::exception& ex)
      {
        res_->set_exception(std::current_exception());
        ec = gpx_parse_error();
      }
      res_->finish_parse_duration(chrono::steady_clock::now() - finish_start);
    }
  };
};
static_assert(bhttp::concepts::body<readable_gpx_body>);
static_assert(bhttp::concepts::body_reader<readable_gpx_body>);

template <bhttp::concepts::body_reader Inner>
struct raw_deflate_body
{
  using value_type = Inner::value_type;

  class reader
  {
    static constexpr auto const chunk_size = 16384;

    Inner::reader r_;
    bool stream_ended_ = false;
    beast::zlib::z_params zs_ = {};
    beast::zlib::inflate_stream inflate_strm_ = {};
    std::unique_ptr<std::array<char, chunk_size>> in_buf_ = nullptr;
    std::unique_ptr<std::array<char, chunk_size>> out_buf_ = nullptr;

    auto inflate_chunk(beast::error_code& ec) -> void
    {
      assert(zs_.avail_in > 0);
      zs_.next_in = in_buf_->data();

      do
      {
        zs_.avail_out = out_buf_->size();
        zs_.next_out = out_buf_->data();

        inflate_strm_.write(zs_, beast::zlib::Flush::none, ec);
        assert(ec != beast::zlib::error::stream_error);
        if (ec == beast::zlib::error::need_buffers)
          ec = {};
        if (ec == beast::zlib::error::end_of_stream)
        {
          ec = {};
          stream_ended_ = true;
        }
        if (ec)
          break;

        auto have = chunk_size - zs_.avail_out;
        auto cbuf =
            net::buffer(static_cast<void const*>(out_buf_->data()), have);

        auto inner_ec = beast::error_code{};
        do
          cbuf += r_.put(cbuf, inner_ec);
        while (cbuf.size() > 0 && !inner_ec);
        if (inner_ec)
        {
          ec = inner_ec;
          break;
        }
      }
      while (zs_.avail_out == 0);

      // In the last case (stream_ended_), the user has sent too much data
      assert(zs_.avail_in == 0 || ec || stream_ended_);
    }

  public:
    template <bool isRequest, bhttp::concepts::fields Fields>
    explicit reader(bhttp::header<isRequest, Fields>& hdr, value_type& v)
      : r_{hdr, v}
    {
    }

    auto
    init(boost::optional<std::uint64_t> /* n */, beast::error_code& ec) noexcept
        -> void
    {
      in_buf_.reset(new std::array<char, chunk_size>{});
      out_buf_.reset(new std::array<char, chunk_size>{});
      ec = {};
    }

    auto
    put(beast::concepts::const_buffer_sequence auto b,
        beast::error_code& ec) noexcept -> std::size_t
    {
      auto in_buf = net::buffer(in_buf_->data(), in_buf_->size());
      in_buf += zs_.avail_in;

      auto consumed_from_b = net::buffer_copy(in_buf, b);
      zs_.avail_in += consumed_from_b;

      if (zs_.avail_in == in_buf_->size())
        inflate_chunk(ec);
      else
        ec = {};

      return consumed_from_b;
    }

    auto finish(beast::error_code& ec) noexcept
    {
      ec = {};
      if (zs_.avail_in > 0)
        inflate_chunk(ec);
      if (!ec && !stream_ended_)
        ec = beast::zlib::error::stream_error;
      if (stream_ended_ && zs_.avail_in > 0)
        // client sent data past deflate stream end
        ec = beast::zlib::error::stream_error;

      in_buf_.reset();
      out_buf_.reset();
      inflate_strm_.reset();
      if (!ec)
      {
        r_.finish(ec);
      }
    }
  };
};

auto handler::handle_process_gpx(l0_ctx ctx, http::readable_request r)
    -> net::awaitable<http::presponse>
{
  auto l =
      l_.sub("handle_process_gpx").with("trace_id", ctx.trace_id.as_string());

  auto gpx_file = gpx::file{};
  try
  {
    l.debug("Reading GPX request body");
    auto const before_read_gpx = chrono::steady_clock::now();
    auto parse_res =
        r.p->get()[bhttp::field::content_encoding] == "x-deflate-raw"
            ? std::move((co_await http::
                             read_request<raw_deflate_body<readable_gpx_body>>(
                                 ctx, std::move(r)))
                            .body())
            : std::move((co_await http::read_request<readable_gpx_body>(
                             ctx, std::move(r)))
                            .body());
    l.debug(
        "Read GPX request body in {}; init took {}, actual parsing {}, "
        "finishing {}",
        chrono::duration<double, std::milli>{
          chrono::steady_clock::now() - before_read_gpx
        },
        chrono::duration<double, std::milli>{parse_res.init_parse_duration()},
        chrono::duration<double, std::milli>{parse_res.put_parse_duration()},
        chrono::duration<double, std::milli>{
          parse_res.finish_parse_duration()
        });
    gpx_file = std::move(parse_res.unwrap());
  }
  catch (std::exception& ex)
  {
    l_.warn("Failed to parse GPX file: {}", ex.what());
    // TODO: more detailed problem reporting
    auto tpl = problem::tpl{
      .status = bhttp::status::bad_request,
      .title = translate("Failed to parse GPX file"),
      .type_uri = "https://routemon.fautchen.eu/problems/gpx-parse-failed",
    };
    co_return http::problem_rsp(
        ctx, tpl.instantiate(), http::keep_alive{false});
  }

  // TODO: catch handler exceptions and return 500 when raised?
  // (keep-alive depends on whether whole request was read)
  auto mres = inner_.process_gpx(ctx.trace_id, std::move(gpx_file));
  if (!mres)
  {
    auto tpl = problem::tpl{
      .status = bhttp::status::internal_server_error,
      .title = translate("Internal server error"),
      .type_uri = "https://routemon.fautchen.eu/problems/"
                  "internal-server-error",
    };
    co_return http::problem_rsp(ctx, tpl.instantiate(), http::keep_alive{true});
  }

  auto rsp = http::make_rsp<bhttp::string_body>(
      bhttp::status::ok, http::keep_alive{true});
  rsp.set(bhttp::field::content_type, "application/json");
  rsp.body() = json::serialize(json::value_from(*mres));
  rsp.prepare_payload();
  co_return rsp;
}

auto handler::handle_sysinfo(l0_ctx ctx, http::readable_request r)
    -> net::awaitable<http::presponse>
{
  auto req = co_await http::read_empty_request(ctx, std::move(r));
  auto info = inner_.sysinfo();

  auto rsp = http::make_rsp<bhttp::string_body>(
      bhttp::status::ok, http::keep_alive{true});
  rsp.set(bhttp::field::content_type, "application/json");
  rsp.body() = json::serialize(json::value_from(info));
  rsp.prepare_payload();
  co_return rsp;
}

handler::handler(log::logger const& l, api::handler&& inner)
  : l_{l.sub("handler")}, inner_{std::move(inner)}
{
}

auto handler::make_routes() -> http::route_tree<http::routed_ctx<outer_ctx>>
{
  auto handler = [this]<class MemFn>(MemFn member)
  { return std::bind_front(member, this); };

  return http::dtree<http::routed_ctx<outer_ctx>>{}.named_subtrees({
    {"gpx", http::dtree<l0_ctx>{{
                                  .post = handler(&handler::handle_process_gpx),
                                }}
                .no_subtrees()},
    {"sysinfo", http::dtree<l0_ctx>{
                  {
                    .get = handler(&handler::handle_sysinfo),
                  }
                }.no_subtrees()},
  });
}

auto server::make_global_middleware(config::http_server const& cfg)
    -> http::middleware_t<http::base_ctx, handler::outer_ctx>
{
  return http::
      middleware_compose<http::base_ctx, http::base_ctx, http::base_ctx>(
          http::expose_content_language_middleware<http::base_ctx>,
          middleware_compose<http::base_ctx, http::base_ctx, http::base_ctx>(
              http::expose_trace_id_middleware<http::base_ctx>,
              http::cors_middleware<http::base_ctx>(cfg.allow_origins)));
}

server::server(
    log::logger const& l, config::http_server const& cfg,
    locale::selector&& lsel, api::handler&& inner)
  : handler_{l, std::move(inner)},
    srv_{
      l,
      http::router{
        l, std::move(lsel), make_global_middleware(cfg), handler_.make_routes()
      }
    }
{
}

auto server::spawn(net::io_context& ioc) -> void { srv_.spawn(ioc); }

} // namespace routemon::srv