ligo/src/udp.cpp

#include <ligo/udp.hpp>

#include <algorithm>
#include <array>
#include <atomic>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <expected>
#include <map>
#include <memory>
#include <optional>
#include <set>
#include <string>
#include <string_view>
#include <utility>
#include <vector>

#include <boost/asio/any_io_executor.hpp>
#include <boost/asio/as_tuple.hpp>
#include <boost/asio/awaitable.hpp>
#include <boost/asio/bind_executor.hpp>
#include <boost/asio/buffer.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/error.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/address.hpp>
#include <boost/asio/ip/udp.hpp>
#include <boost/asio/post.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/strand.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/json/object.hpp>
#include <boost/json/parse.hpp>
#include <boost/json/serialize.hpp>
#include <boost/json/value.hpp>
#include <boost/system/error_code.hpp>
#include <openssl/err.h>
#include <openssl/ssl.h>
#include <openssl/x509_vfy.h>

#include <modulus.hpp>

#include <ligo/detail/errors.hpp>
#include <ligo/detail/state.hpp>
#include <ligo/detail/transport.hpp>
#include <ligo/config.hpp>
#include <ligo/security.hpp>

namespace {

struct ssl_ctx_deleter {
  auto operator()(SSL_CTX* context) const noexcept -> void;
};

struct ssl_deleter {
  auto operator()(SSL* ssl) const noexcept -> void;
};

struct bio_deleter {
  auto operator()(BIO* bio) const noexcept -> void;
};

struct dtls_state {
  std::unique_ptr<SSL_CTX, ssl_ctx_deleter> context{};
  std::unique_ptr<SSL, ssl_deleter> ssl{};
  std::unique_ptr<BIO, bio_deleter> network_bio{};
  std::bool_t server{};
  std::bool_t handshake_started{};
  std::bool_t handshake_complete{};
};

struct dtls_processed {
  std::vector<std::string> wire{};
  std::vector<std::string> plaintext{};
};

struct reliable_channel_state {
  reliable_channel_state(std::unique_ptr<boost::asio::ip::udp::socket> socket, boost::asio::ip::udp::endpoint local_endpoint, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::unique_ptr<dtls_state> dtls);
  std::unique_ptr<boost::asio::ip::udp::socket> socket{};
  boost::asio::strand<boost::asio::ip::udp::socket::executor_type> strand{boost::asio::make_strand(socket->get_executor())};
  boost::asio::ip::udp::endpoint local_endpoint{};
  boost::asio::ip::udp::endpoint peer{};
  ligo::udp_reliability_config config{};
  ligo::security_config security{};
  std::unique_ptr<dtls_state> dtls{};
  std::uint64_t session_id{1};
  std::uint64_t next_send_sequence{1};
  std::uint64_t next_receive_sequence{1};
  std::uint32_t in_flight{};
  std::uint32_t active_operations{};
  std::array<std::char_t, 65536> receive_buffer{};
  boost::asio::ip::udp::endpoint receive_sender{};
  std::bool_t receive_loop_started{};
  std::atomic<std::bool_t> close_requested{};
  std::bool_t closed{};
  std::bool_t receive_waiting{};
  boost::asio::steady_timer notify_timer{strand};
  boost::asio::steady_timer close_timer{strand};
  std::optional<modulus::error> receive_error{};
  std::set<std::uint64_t> active_send_sequences{};
  std::set<std::uint64_t> acked{};
  std::map<std::uint64_t, std::weak_ptr<boost::asio::steady_timer>> ack_waiters{};
  std::map<std::uint64_t, boost::json::value> pending{};
};

constexpr auto max_datagram_size = std::size_t{65507};
constexpr auto max_dtls_plaintext_size = std::size_t{12000};

auto same_peer(boost::asio::ip::udp::endpoint const& left, boost::asio::ip::udp::endpoint const& right) -> std::bool_t {
  return left.address() == right.address() && left.port() == right.port();
}

auto uint_member(boost::json::object const& object, std::string_view name) -> std::optional<std::uint64_t> {
  auto found = object.find(name);
  if (found == object.end() || !found->value().is_number()) {
    return std::nullopt;
  }
  if (found->value().is_int64() && found->value().as_int64() < 0) {
    return std::nullopt;
  }
  if (found->value().is_double()) {
    return std::nullopt;
  }
  return found->value().is_uint64() ? found->value().as_uint64() : static_cast<std::uint64_t>(found->value().as_int64());
}

auto string_member(boost::json::object const& object, std::string_view name) -> std::string {
  auto found = object.find(name);
  if (found == object.end() || !found->value().is_string()) {
    return {};
  }
  return std::string{found->value().as_string()};
}

auto make_data(std::uint64_t session_id, std::uint64_t sequence, boost::json::value payload) -> std::string {
  auto envelope = boost::json::object{};
  envelope["kind"] = "data";
  envelope["session"] = session_id;
  envelope["sequence"] = sequence;
  envelope["payload"] = std::move(payload);
  return boost::json::serialize(envelope);
}

auto make_ack(std::uint64_t session_id, std::uint64_t sequence) -> std::string {
  auto envelope = boost::json::object{};
  envelope["kind"] = "ack";
  envelope["session"] = session_id;
  envelope["sequence"] = sequence;
  return boost::json::serialize(envelope);
}

auto validate_config(ligo::udp_reliability_config const& config) -> std::expected<void, modulus::error> {
  if (config.resend_interval <= std::chrono::milliseconds{0}) {
    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP resend_interval must be positive"));
  }
  if (config.max_in_flight == 0) {
    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP max_in_flight must be positive"));
  }
  if (config.session_timeout <= std::chrono::milliseconds{0}) {
    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP session_timeout must be positive"));
  }
  return {};
}

auto dtls_failure(std::int32_t code, std::string message) -> modulus::error {
  auto const error = ERR_get_error();
  if (error != 0) {
    message += ": ";
    message += ERR_error_string(error, nullptr);
  }
  return ligo::detail::make_error(code, std::move(message));
}

auto dtls_failure(std::string message) -> modulus::error {
  return dtls_failure(ligo::detail::error_unavailable, std::move(message));
}

auto validate_security(ligo::security_config const& config, std::bool_t server) -> std::expected<void, modulus::error> {
  if (config.mode == ligo::security_mode::TLS) {
    return ligo::detail::fail_void(ligo::unsupported_security("UDP", config.mode));
  }
  if (config.mode != ligo::security_mode::DTLS) {
    return {};
  }
  if (config.tls_context) {
    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP DTLS does not accept a TLS context"));
  }
  if (server && (config.certificate_chain_file.empty() || config.private_key_file.empty())) {
    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP DTLS server requires certificate chain/private key files"));
  }
  if (!config.certificate_chain_file.empty() != !config.private_key_file.empty()) {
    return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP DTLS certificate chain and private key must both be configured"));
  }
  return {};
}

auto load_dtls_verify_paths(SSL_CTX* context, ligo::security_config const& config) -> std::expected<void, modulus::error> {
  if (config.verify == ligo::verify_mode::NONE) {
    return {};
  }
  if (!config.ca_file.empty()) {
    if (SSL_CTX_load_verify_locations(context, config.ca_file.c_str(), nullptr) != 1) {
      return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS CA file"));
    }
    return {};
  }
  if (SSL_CTX_set_default_verify_paths(context) != 1) {
    return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS default verify paths"));
  }
  return {};
}

auto load_dtls_identity(SSL_CTX* context, ligo::security_config const& config) -> std::expected<void, modulus::error> {
  if (config.certificate_chain_file.empty() && config.private_key_file.empty()) {
    return {};
  }
  if (SSL_CTX_use_certificate_chain_file(context, config.certificate_chain_file.c_str()) != 1) {
    return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS certificate chain"));
  }
  if (SSL_CTX_use_PrivateKey_file(context, config.private_key_file.c_str(), SSL_FILETYPE_PEM) != 1) {
    return ligo::detail::fail_void(dtls_failure("failed to load UDP DTLS private key"));
  }
  if (SSL_CTX_check_private_key(context) != 1) {
    return ligo::detail::fail_void(dtls_failure(ligo::detail::error_invalid_argument, "UDP DTLS private key does not match certificate"));
  }
  return {};
}

auto make_dtls_state(ligo::security_config const& config, std::bool_t server) -> std::expected<std::unique_ptr<dtls_state>, modulus::error> {
  if (config.mode != ligo::security_mode::DTLS) {
    return modulus::success(std::unique_ptr<dtls_state>{});
  }
  auto valid = validate_security(config, server);
  if (!valid) {
    return ligo::detail::fail<std::unique_ptr<dtls_state>>(valid.error());
  }
  auto context = std::unique_ptr<SSL_CTX, ssl_ctx_deleter>{SSL_CTX_new(server ? DTLS_server_method() : DTLS_client_method())};
  if (!context) {
    return ligo::detail::fail<std::unique_ptr<dtls_state>>(ligo::detail::error_unavailable, "failed to create UDP DTLS context");
  }
  SSL_CTX_set_options(context.get(), SSL_OP_NO_COMPRESSION | SSL_OP_NO_QUERY_MTU);
  if (SSL_CTX_set_min_proto_version(context.get(), DTLS1_2_VERSION) != 1) {
    return ligo::detail::fail<std::unique_ptr<dtls_state>>(dtls_failure("failed to configure UDP DTLS minimum protocol"));
  }
  auto const verify_peer = config.verify == ligo::verify_mode::PEER || (!server && config.verify == ligo::verify_mode::DEFAULTED);
  auto const verify = verify_peer ? SSL_VERIFY_PEER | (server ? SSL_VERIFY_FAIL_IF_NO_PEER_CERT : 0) : SSL_VERIFY_NONE;
  SSL_CTX_set_verify(context.get(), verify, nullptr);
  auto verify_paths = load_dtls_verify_paths(context.get(), config);
  if (!verify_paths) {
    return ligo::detail::fail<std::unique_ptr<dtls_state>>(verify_paths.error());
  }
  auto identity = load_dtls_identity(context.get(), config);
  if (!identity) {
    return ligo::detail::fail<std::unique_ptr<dtls_state>>(identity.error());
  }
  auto ssl = std::unique_ptr<SSL, ssl_deleter>{SSL_new(context.get())};
  if (!ssl) {
    return ligo::detail::fail<std::unique_ptr<dtls_state>>(ligo::detail::error_unavailable, "failed to create UDP DTLS session");
  }
  if (!server && verify_peer && !config.peer_name.empty()) {
    if (X509_VERIFY_PARAM_set1_host(SSL_get0_param(ssl.get()), config.peer_name.c_str(), 0) != 1) {
      return ligo::detail::fail<std::unique_ptr<dtls_state>>(dtls_failure("failed to configure UDP DTLS peer identity"));
    }
  }
  SSL_set_mtu(ssl.get(), 1200);
  auto* ssl_bio = static_cast<BIO*>(nullptr);
  auto* network_bio = static_cast<BIO*>(nullptr);
  if (BIO_new_bio_dgram_pair(&ssl_bio, 0, &network_bio, 0) != 1 || ssl_bio == nullptr || network_bio == nullptr) {
    BIO_free(ssl_bio);
    BIO_free(network_bio);
    return ligo::detail::fail<std::unique_ptr<dtls_state>>(dtls_failure("failed to create UDP DTLS BIO pair"));
  }
  BIO_up_ref(ssl_bio);
  SSL_set_bio(ssl.get(), ssl_bio, ssl_bio);
  if (server) {
    SSL_set_accept_state(ssl.get());
  } else {
    SSL_set_connect_state(ssl.get());
  }
  return modulus::success(std::make_unique<dtls_state>(dtls_state{
    .context = std::move(context),
    .ssl = std::move(ssl),
    .network_bio = std::unique_ptr<BIO, bio_deleter>{network_bio},
    .server = server,
  }));
}

auto drain_dtls_output(dtls_state& state) -> std::expected<std::vector<std::string>, modulus::error> {
  auto output = std::vector<std::string>{};
  auto* bio = state.network_bio.get();
  while (BIO_ctrl_pending(bio) > 0) {
    auto const pending = static_cast<std::size_t>(BIO_ctrl_pending(bio));
    auto datagram = std::string(std::min(pending, max_datagram_size), '\0');
    auto const read = BIO_read(bio, datagram.data(), static_cast<std::int32_t>(datagram.size()));
    if (read <= 0) {
      return ligo::detail::fail<std::vector<std::string>>(dtls_failure("failed to read UDP DTLS output"));
    }
    datagram.resize(static_cast<std::size_t>(read));
    output.push_back(std::move(datagram));
  }
  return modulus::success(std::move(output));
}

auto append_dtls_output(std::vector<std::string>& target, dtls_state& state) -> std::expected<void, modulus::error> {
  auto drained = drain_dtls_output(state);
  if (!drained) {
    return ligo::detail::fail_void(drained.error());
  }
  for (auto& datagram : *drained) {
    target.push_back(std::move(datagram));
  }
  return {};
}

auto run_dtls_handshake(dtls_state& state, std::vector<std::string>& output) -> std::expected<void, modulus::error> {
  if (state.handshake_complete) {
    return {};
  }
  state.handshake_started = true;
  auto const result = SSL_do_handshake(state.ssl.get());
  if (result == 1) {
    state.handshake_complete = true;
    return append_dtls_output(output, state);
  }
  auto const error = SSL_get_error(state.ssl.get(), result);
  if (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE) {
    return append_dtls_output(output, state);
  }
  return ligo::detail::fail_void(dtls_failure("UDP DTLS handshake failed"));
}

auto protect_dtls_payload(reliable_channel_state& state, std::string_view plaintext) -> std::expected<std::vector<std::string>, modulus::error> {
  if (!state.dtls) {
    return modulus::success(std::vector<std::string>{
      std::string{plaintext},
    });
  }
  if (plaintext.size() > max_dtls_plaintext_size) {
    return ligo::detail::fail<std::vector<std::string>>(ligo::detail::error_invalid_argument, "UDP DTLS payload is too large");
  }
  auto output = std::vector<std::string>{};
  auto handshake = run_dtls_handshake(*state.dtls, output);
  if (!handshake) {
    return ligo::detail::fail<std::vector<std::string>>(handshake.error());
  }
  if (!state.dtls->handshake_complete) {
    return modulus::success(std::move(output));
  }
  auto const written = SSL_write(state.dtls->ssl.get(), plaintext.data(), static_cast<std::int32_t>(plaintext.size()));
  if (written <= 0) {
    auto const error = SSL_get_error(state.dtls->ssl.get(), written);
    if (error != SSL_ERROR_WANT_READ && error != SSL_ERROR_WANT_WRITE) {
      return ligo::detail::fail<std::vector<std::string>>(dtls_failure("UDP DTLS write failed"));
    }
  }
  auto drained = append_dtls_output(output, *state.dtls);
  if (!drained) {
    return ligo::detail::fail<std::vector<std::string>>(drained.error());
  }
  return modulus::success(std::move(output));
}

auto process_dtls_input(reliable_channel_state& state, std::string_view datagram) -> std::expected<dtls_processed, modulus::error> {
  auto processed = dtls_processed{};
  if (!state.dtls) {
    processed.plaintext.push_back(std::string{datagram});
    return modulus::success(std::move(processed));
  }
  auto const written = BIO_write(state.dtls->network_bio.get(), datagram.data(), static_cast<std::int32_t>(datagram.size()));
  if (written <= 0) {
    return ligo::detail::fail<dtls_processed>(dtls_failure("failed to write UDP DTLS input"));
  }
  auto handshake = run_dtls_handshake(*state.dtls, processed.wire);
  if (!handshake) {
    return ligo::detail::fail<dtls_processed>(handshake.error());
  }
  if (!state.dtls->handshake_complete) {
    return modulus::success(std::move(processed));
  }
  for (;;) {
    auto plaintext = std::string(max_dtls_plaintext_size, '\0');
    auto const read = SSL_read(state.dtls->ssl.get(), plaintext.data(), static_cast<std::int32_t>(plaintext.size()));
    if (read > 0) {
      plaintext.resize(static_cast<std::size_t>(read));
      processed.plaintext.push_back(std::move(plaintext));
      continue;
    }
    auto const error = SSL_get_error(state.dtls->ssl.get(), read);
    if (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE || error == SSL_ERROR_ZERO_RETURN) {
      break;
    }
    return ligo::detail::fail<dtls_processed>(dtls_failure("UDP DTLS read failed"));
  }
  auto drained = append_dtls_output(processed.wire, *state.dtls);
  if (!drained) {
    return ligo::detail::fail<dtls_processed>(drained.error());
  }
  return modulus::success(std::move(processed));
}

auto channel_state(ligo::detail::state_handle const& handle) -> std::shared_ptr<reliable_channel_state> {
  return handle.as<reliable_channel_state>();
}

auto notify_waiters(std::shared_ptr<reliable_channel_state> const& state) -> void {
  state->notify_timer.cancel();
}

auto notify_close_waiters(std::shared_ptr<reliable_channel_state> const& state) -> void {
  state->close_timer.cancel();
}

auto notify_send_waiters(std::shared_ptr<reliable_channel_state> const& state) -> void {
  for (auto& [_, waiter] : state->ack_waiters) {
    if (auto timer = waiter.lock()) {
      timer->cancel();
    }
  }
}

auto send_wire_datagrams(std::shared_ptr<reliable_channel_state> const& state, std::vector<std::string> const& datagrams) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  for (auto const& datagram : datagrams) {
    auto [send_error, _] = co_await state->socket->async_send_to(boost::asio::buffer(datagram), state->peer, boost::asio::bind_executor(state->strand, boost::asio::as_tuple(boost::asio::use_awaitable)));
    if (send_error) {
      co_return ligo::detail::fail_void(ligo::detail::network_failure(send_error));
    }
  }
  co_return std::expected<void, modulus::error>{};
}

auto send_ack(std::shared_ptr<reliable_channel_state> const& state, std::uint64_t sequence) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  auto ack_body = make_ack(state->session_id, sequence);
  auto datagrams = protect_dtls_payload(*state, ack_body);
  if (!datagrams) {
    co_return ligo::detail::fail_void(datagrams.error());
  }
  co_return co_await send_wire_datagrams(state, *datagrams);
}

auto close_state(std::shared_ptr<reliable_channel_state> const& state) -> void {
  if (!state || state->closed) {
    return;
  }
  state->close_requested.store(true, std::memory_order_release);
  state->closed = true;
  state->receive_error = ligo::detail::unavailable("UDP channel closed");
  auto ignored = boost::system::error_code{};
  if (state->socket && state->socket->is_open()) {
    state->socket->cancel(ignored);
    state->socket->close(ignored);
  }
  notify_send_waiters(state);
  notify_waiters(state);
  notify_close_waiters(state);
}

auto receive_loop(std::shared_ptr<reliable_channel_state> state) -> boost::asio::awaitable<void> {
  for (;;) {
    if (state->active_operations == 0) {
      state->receive_loop_started = false;
      notify_close_waiters(state);
      co_return;
    }
    auto [receive_error, size] = co_await state->socket->async_receive_from(boost::asio::buffer(state->receive_buffer), state->receive_sender, boost::asio::bind_executor(state->strand, boost::asio::as_tuple(boost::asio::use_awaitable)));
    if (receive_error) {
      if (state->closed) {
        state->receive_loop_started = false;
        notify_close_waiters(state);
        co_return;
      }
      if (receive_error == boost::asio::error::operation_aborted) {
        if (state->active_operations == 0) {
          state->receive_loop_started = false;
          notify_close_waiters(state);
          co_return;
        }
        continue;
      }
      state->receive_error = ligo::detail::network_failure(receive_error);
      notify_send_waiters(state);
      notify_waiters(state);
      continue;
    }
    if (size > max_datagram_size || !same_peer(state->receive_sender, state->peer)) {
      continue;
    }
    auto processed = process_dtls_input(*state, std::string_view{
      state->receive_buffer.data(),
      size,
    });
    if (!processed) {
      state->receive_error = processed.error();
      notify_send_waiters(state);
      notify_waiters(state);
      continue;
    }
    auto sent_dtls = co_await send_wire_datagrams(state, processed->wire);
    if (!sent_dtls) {
      state->receive_error = sent_dtls.error();
      notify_send_waiters(state);
      notify_waiters(state);
      continue;
    }
    for (auto& plaintext : processed->plaintext) {
      auto parse_error = boost::system::error_code{};
      auto envelope = boost::json::parse(plaintext, parse_error);
      if (parse_error || !envelope.is_object()) {
        continue;
      }
      auto const& object = envelope.as_object();
      auto session = uint_member(object, "session");
      auto sequence = uint_member(object, "sequence");
      if (!session.has_value() || *session != state->session_id || !sequence.has_value()) {
        continue;
      }
      auto const kind = string_member(object, "kind");
      if (kind == "ack") {
        if (state->active_send_sequences.contains(*sequence)) {
          state->acked.insert(*sequence);
          if (auto found = state->ack_waiters.find(*sequence); found != state->ack_waiters.end()) {
            if (auto timer = found->second.lock()) {
              timer->cancel();
            }
          }
          notify_waiters(state);
        }
        continue;
      }
      if (kind != "data") {
        continue;
      }
      auto payload = object.find("payload");
      if (payload == object.end()) {
        continue;
      }
      if (*sequence < state->next_receive_sequence) {
        auto ack = co_await send_ack(state, *sequence);
        if (!ack) {
          state->receive_error = ack.error();
          notify_send_waiters(state);
          notify_waiters(state);
        }
        continue;
      }
      if (*sequence - state->next_receive_sequence >= state->config.max_in_flight) {
        continue;
      }
      auto ack = co_await send_ack(state, *sequence);
      if (!ack) {
        state->receive_error = ack.error();
        notify_send_waiters(state);
        notify_waiters(state);
        continue;
      }
      if (state->pending.contains(*sequence)) {
        continue;
      }
      state->pending.emplace(*sequence, std::move(payload->value()));
      notify_waiters(state);
    }
  }
}

auto ensure_receive_loop(std::shared_ptr<reliable_channel_state> const& state) -> void {
  if (state->receive_loop_started || state->closed) {
    return;
  }
  state->receive_loop_started = true;
  boost::asio::co_spawn(state->strand, receive_loop(state), boost::asio::detached);
}

auto begin_operation(std::shared_ptr<reliable_channel_state> const& state) -> std::expected<void, modulus::error> {
  if (state->closed || state->close_requested.load(std::memory_order_acquire)) {
    return ligo::detail::fail_void(ligo::detail::unavailable("UDP channel closed"));
  }
  ++state->active_operations;
  ensure_receive_loop(state);
  return {};
}

auto finish_operation(std::shared_ptr<reliable_channel_state> const& state) -> void {
  if (state->active_operations > 0) {
    --state->active_operations;
  }
  if (state->active_operations == 0) {
    auto ignored = boost::system::error_code{};
    if (state->socket && state->socket->is_open()) {
      state->socket->cancel(ignored);
    }
    notify_close_waiters(state);
  }
}

auto finish_send(std::shared_ptr<reliable_channel_state> const& state, std::uint64_t sequence) -> void {
  if (state->in_flight > 0) {
    --state->in_flight;
  }
  state->active_send_sequences.erase(sequence);
  state->acked.erase(sequence);
  state->ack_waiters.erase(sequence);
  finish_operation(state);
}

auto udp_endpoint_from_parts(ligo::detail::endpoint_parts const& parts) -> std::expected<boost::asio::ip::udp::endpoint, modulus::error> {
  auto error = boost::system::error_code{};
  auto address = boost::asio::ip::make_address(parts.host, error);
  if (error) {
    return ligo::detail::fail<boost::asio::ip::udp::endpoint>(ligo::detail::error_unavailable, error.message());
  }
  return modulus::success(boost::asio::ip::udp::endpoint{
    address,
    parts.port,
  });
}

auto open_state(boost::asio::any_io_executor executor, boost::asio::ip::udp::endpoint local, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::bool_t server) -> std::expected<std::shared_ptr<reliable_channel_state>, modulus::error> {
  auto valid = validate_config(config);
  if (!valid) {
    return ligo::detail::fail<std::shared_ptr<reliable_channel_state>>(valid.error());
  }
  auto dtls = make_dtls_state(security, server);
  if (!dtls) {
    return ligo::detail::fail<std::shared_ptr<reliable_channel_state>>(dtls.error());
  }
  auto socket = std::make_unique<boost::asio::ip::udp::socket>(std::move(executor));
  auto error = boost::system::error_code{};
  socket->open(local.protocol(), error);
  if (error) {
    auto failure_status = ligo::detail::network_failure(error);
    return ligo::detail::fail<std::shared_ptr<reliable_channel_state>>(failure_status);
  }
  socket->bind(local, error);
  if (error) {
    auto failure_status = ligo::detail::network_failure(error);
    return ligo::detail::fail<std::shared_ptr<reliable_channel_state>>(failure_status);
  }
  auto local_endpoint = socket->local_endpoint(error);
  if (error) {
    auto failure_status = ligo::detail::network_failure(error);
    return ligo::detail::fail<std::shared_ptr<reliable_channel_state>>(failure_status);
  }
  return modulus::success(std::make_shared<reliable_channel_state>(std::move(socket), local_endpoint, std::move(peer), config, std::move(security), session_id, std::move(*dtls)));
}

auto adopt_state(std::unique_ptr<boost::asio::ip::udp::socket> socket, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::bool_t server) -> std::expected<std::shared_ptr<reliable_channel_state>, modulus::error> {
  auto valid = validate_config(config);
  if (!valid) {
    return ligo::detail::fail<std::shared_ptr<reliable_channel_state>>(valid.error());
  }
  auto dtls = make_dtls_state(security, server);
  if (!dtls) {
    return ligo::detail::fail<std::shared_ptr<reliable_channel_state>>(dtls.error());
  }
  if (!socket) {
    return ligo::detail::fail<std::shared_ptr<reliable_channel_state>>(ligo::detail::error_invalid_argument, "UDP socket is null");
  }
  if (!socket->is_open()) {
    return ligo::detail::fail<std::shared_ptr<reliable_channel_state>>(ligo::detail::error_invalid_argument, "UDP socket is not open");
  }
  auto error = boost::system::error_code{};
  auto local_endpoint = socket->local_endpoint(error);
  if (error || local_endpoint.port() == 0) {
    return ligo::detail::fail<std::shared_ptr<reliable_channel_state>>(ligo::detail::error_invalid_argument, "UDP socket is not bound");
  }
  return modulus::success(std::make_shared<reliable_channel_state>(std::move(socket), local_endpoint, std::move(peer), config, std::move(security), session_id, std::move(*dtls)));
}

auto send_payload_on_strand(std::shared_ptr<reliable_channel_state> state, boost::json::value payload) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  if (state->in_flight >= state->config.max_in_flight) {
    co_return ligo::detail::fail_void(ligo::detail::unavailable("too many UDP messages in flight"));
  }
  auto const sequence = state->next_send_sequence;
  auto body = make_data(state->session_id, sequence, std::move(payload));
  if (body.size() > max_datagram_size) {
    co_return ligo::detail::fail_void(ligo::detail::invalid_argument("UDP payload is too large"));
  }
  auto begin = begin_operation(state);
  if (!begin) {
    co_return begin;
  }
  ++state->next_send_sequence;
  ++state->in_flight;
  state->active_send_sequences.insert(sequence);
  auto timer = std::make_shared<boost::asio::steady_timer>(state->strand);
  auto deadline = std::chrono::steady_clock::now() + state->config.session_timeout;
  for (;;) {
    if (state->closed || state->close_requested.load(std::memory_order_acquire)) {
      finish_send(state, sequence);
      co_return ligo::detail::fail_void(ligo::detail::unavailable("UDP channel closed"));
    }
    auto datagrams = protect_dtls_payload(*state, body);
    if (!datagrams) {
      finish_send(state, sequence);
      co_return ligo::detail::fail_void(datagrams.error());
    }
    auto sent = co_await send_wire_datagrams(state, *datagrams);
    if (state->closed || state->close_requested.load(std::memory_order_acquire)) {
      finish_send(state, sequence);
      co_return ligo::detail::fail_void(ligo::detail::unavailable("UDP channel closed"));
    }
    if (!sent) {
      finish_send(state, sequence);
      co_return ligo::detail::fail_void(sent.error());
    }

    if (state->acked.erase(sequence) > 0) {
      finish_send(state, sequence);
      co_return std::expected<void, modulus::error>{};
    }
    if (state->receive_error.has_value()) {
      auto receive_status = *state->receive_error;
      state->receive_error.reset();
      finish_send(state, sequence);
      co_return ligo::detail::fail_void(receive_status);
    }
    auto now = std::chrono::steady_clock::now();
    if (now >= deadline) {
      finish_send(state, sequence);
      co_return ligo::detail::fail_void(ligo::detail::error_timeout, "timed out waiting for UDP ACK");
    }
    auto wake_at = now + state->config.resend_interval;
    if (deadline < wake_at) {
      wake_at = deadline;
    }
    timer->expires_at(wake_at);
    state->ack_waiters[sequence] = timer;
    auto [timer_error] = co_await timer->async_wait(boost::asio::as_tuple(boost::asio::use_awaitable));
    state->ack_waiters.erase(sequence);
    if (state->closed || state->close_requested.load(std::memory_order_acquire)) {
      finish_send(state, sequence);
      co_return ligo::detail::fail_void(ligo::detail::unavailable("UDP channel closed"));
    }
    if (timer_error && timer_error != boost::asio::error::operation_aborted) {
      finish_send(state, sequence);
      co_return ligo::detail::fail_void(ligo::detail::network_failure(timer_error));
    }
    if (state->acked.erase(sequence) > 0) {
      finish_send(state, sequence);
      co_return std::expected<void, modulus::error>{};
    }
    if (state->receive_error.has_value()) {
      auto receive_status = *state->receive_error;
      state->receive_error.reset();
      finish_send(state, sequence);
      co_return ligo::detail::fail_void(receive_status);
    }
  }
}

auto send_payload(std::shared_ptr<reliable_channel_state> state, boost::json::value payload) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  if (!state) {
    co_return ligo::detail::fail_void(ligo::detail::unavailable("UDP channel is not initialized"));
  }
  co_return co_await boost::asio::co_spawn(state->strand, send_payload_on_strand(state, std::move(payload)), boost::asio::use_awaitable);
}

auto receive_payload_on_strand(std::shared_ptr<reliable_channel_state> state) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
  if (state->receive_waiting) {
    co_return ligo::detail::fail<boost::json::value>(ligo::detail::error_unavailable, "UDP receive is already pending");
  }
  auto begin = begin_operation(state);
  if (!begin) {
    co_return ligo::detail::fail<boost::json::value>(begin.error());
  }
  state->receive_waiting = true;
  for (;;) {
    if (state->closed || state->close_requested.load(std::memory_order_acquire)) {
      state->receive_waiting = false;
      finish_operation(state);
      co_return ligo::detail::fail<boost::json::value>(ligo::detail::error_unavailable, "UDP channel closed");
    }
    if (auto found = state->pending.find(state->next_receive_sequence); found != state->pending.end()) {
      auto value = std::move(found->second);
      state->pending.erase(found);
      ++state->next_receive_sequence;
      state->receive_waiting = false;
      finish_operation(state);
      co_return modulus::success(std::move(value));
    }
    if (state->receive_error.has_value()) {
      auto receive_status = *state->receive_error;
      state->receive_error.reset();
      state->receive_waiting = false;
      finish_operation(state);
      co_return ligo::detail::fail<boost::json::value>(receive_status);
    }
    auto [timer_error] = co_await state->notify_timer.async_wait(boost::asio::as_tuple(boost::asio::use_awaitable));
    if (timer_error && timer_error != boost::asio::error::operation_aborted) {
      state->receive_waiting = false;
      finish_operation(state);
      auto failure_status = ligo::detail::network_failure(timer_error);
      co_return ligo::detail::fail<boost::json::value>(failure_status);
    }
  }
}

auto receive_payload(std::shared_ptr<reliable_channel_state> state) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
  if (!state) {
    co_return ligo::detail::fail<boost::json::value>(ligo::detail::error_unavailable, "UDP channel is not initialized");
  }
  co_return co_await boost::asio::co_spawn(state->strand, receive_payload_on_strand(state), boost::asio::use_awaitable);
}

auto receive_message(std::shared_ptr<reliable_channel_state> state) -> boost::asio::awaitable<std::expected<ligo::udp::message, modulus::error>> {
  auto payload = co_await receive_payload(std::move(state));
  if (!payload) {
    co_return ligo::detail::fail<ligo::udp::message>(payload.error());
  }
  co_return modulus::success(ligo::udp::message{
    .payload = std::move(*payload),
  });
}

auto local_endpoint_for(std::shared_ptr<reliable_channel_state> const& state) -> std::expected<boost::asio::ip::udp::endpoint, modulus::error> {
  if (!state || !state->socket) {
    return ligo::detail::fail<boost::asio::ip::udp::endpoint>(ligo::detail::error_unavailable, "UDP channel is not initialized");
  }
  return modulus::success(state->local_endpoint);
}

auto close_channel(std::shared_ptr<reliable_channel_state> state) -> void {
  if (!state) {
    return;
  }
  state->close_requested.store(true, std::memory_order_release);
  boost::asio::post(state->strand, [state] {
    close_state(state);
  });
}

auto channel_drained(std::shared_ptr<reliable_channel_state> const& state) -> std::bool_t {
  return !state || (state->active_operations == 0 && !state->receive_loop_started);
}

auto async_close_channel_on_strand(std::shared_ptr<reliable_channel_state> state, std::chrono::milliseconds timeout) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  close_state(state);
  if (channel_drained(state)) {
    co_return std::expected<void, modulus::error>{};
  }
  auto deadline_timer = std::make_shared<boost::asio::steady_timer>(state->strand);
  auto timed_out = std::make_shared<std::atomic<std::bool_t>>(false);
  deadline_timer->expires_after(timeout);
  deadline_timer->async_wait([state, deadline_timer, timed_out](boost::system::error_code const& error) {
    if (!error) {
      timed_out->store(true, std::memory_order_release);
      notify_close_waiters(state);
    }
  });
  for (;;) {
    if (channel_drained(state)) {
      deadline_timer->cancel();
      co_return std::expected<void, modulus::error>{};
    }
    if (timed_out->load(std::memory_order_acquire)) {
      co_return ligo::detail::fail_void(ligo::detail::error_timeout, "timed out waiting for UDP channel to drain");
    }
    state->close_timer.expires_at(std::chrono::steady_clock::time_point::max());
    auto [timer_error] = co_await state->close_timer.async_wait(boost::asio::as_tuple(boost::asio::use_awaitable));
    if (timer_error && timer_error != boost::asio::error::operation_aborted) {
      deadline_timer->cancel();
      co_return ligo::detail::fail_void(ligo::detail::network_failure(timer_error));
    }
  }
}

auto async_close_channel(std::shared_ptr<reliable_channel_state> state, std::chrono::milliseconds timeout) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  if (!state) {
    co_return std::expected<void, modulus::error>{};
  }
  auto valid_timeout = ligo::detail::validate_positive_timeout(timeout, "UDP shutdown timeout");
  if (!valid_timeout) {
    co_return valid_timeout;
  }
  co_return co_await boost::asio::co_spawn(state->strand, async_close_channel_on_strand(state, timeout), boost::asio::use_awaitable);
}

auto endpoint_from_config(ligo::endpoint_config const& config) -> std::expected<ligo::udp::endpoint, modulus::error> {
  auto local_parts = ligo::detail::parse_server_endpoint_text(config.bind);
  if (!local_parts) {
    return ligo::detail::fail<ligo::udp::endpoint>(local_parts.error());
  }
  auto peer_parts = ligo::detail::parse_client_endpoint_text(config.connect);
  if (!peer_parts) {
    return ligo::detail::fail<ligo::udp::endpoint>(peer_parts.error());
  }
  auto local = udp_endpoint_from_parts(*local_parts);
  if (!local) {
    return ligo::detail::fail<ligo::udp::endpoint>(local.error());
  }
  auto peer = udp_endpoint_from_parts(*peer_parts);
  if (!peer) {
    return ligo::detail::fail<ligo::udp::endpoint>(peer.error());
  }
  return modulus::success(ligo::udp::endpoint{
    .local = *local,
    .peer = *peer,
  });
}

auto ssl_ctx_deleter::operator()(SSL_CTX* context) const noexcept -> void {
  SSL_CTX_free(context);
}

auto ssl_deleter::operator()(SSL* ssl) const noexcept -> void {
  SSL_free(ssl);
}

auto bio_deleter::operator()(BIO* bio) const noexcept -> void {
  BIO_free(bio);
}

reliable_channel_state::reliable_channel_state(std::unique_ptr<boost::asio::ip::udp::socket> socket, boost::asio::ip::udp::endpoint local_endpoint, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config, ligo::security_config security, std::uint64_t session_id, std::unique_ptr<dtls_state> dtls) : socket{std::move(socket)}, local_endpoint{std::move(local_endpoint)}, peer{std::move(peer)}, config{config}, security{std::move(security)}, dtls{std::move(dtls)}, session_id{session_id} {
  notify_timer.expires_at(std::chrono::steady_clock::time_point::max());
  close_timer.expires_at(std::chrono::steady_clock::time_point::max());
}

}

ligo::udp::client::~client() {
  close();
}

auto ligo::udp::client::operator=(ligo::udp::client&& other) noexcept -> ligo::udp::client& {
  if (this != &other) {
    close();
    state_ = std::move(other.state_);
  }
  return *this;
}

auto ligo::udp::client::open(boost::asio::any_io_executor executor, boost::asio::ip::udp::endpoint peer, ligo::udp::client_options options) -> std::expected<ligo::udp::client, modulus::error> {
  return ligo::udp::client::bind(std::move(executor), {
    .local = {
      peer.protocol(),
      0,
    },
    .peer = std::move(peer),
  }, options);
}

auto ligo::udp::client::open(boost::asio::io_context& io, boost::asio::ip::udp::endpoint peer, ligo::udp::client_options options) -> std::expected<ligo::udp::client, modulus::error> {
  return ligo::udp::client::open(io.get_executor(), std::move(peer), options);
}

auto ligo::udp::client::bind(boost::asio::any_io_executor executor, ligo::udp::endpoint endpoint, ligo::udp::client_options options) -> std::expected<ligo::udp::client, modulus::error> {
  auto state = open_state(std::move(executor), std::move(endpoint.local), std::move(endpoint.peer), options.reliability, options.security, options.session_id, false);
  if (!state) {
    return ligo::detail::fail<ligo::udp::client>(state.error());
  }
  return modulus::success(ligo::udp::client{
    ligo::detail::state_handle{std::move(*state)},
  });
}

auto ligo::udp::client::bind(boost::asio::io_context& io, ligo::udp::endpoint endpoint, ligo::udp::client_options options) -> std::expected<ligo::udp::client, modulus::error> {
  return ligo::udp::client::bind(io.get_executor(), std::move(endpoint), options);
}

auto ligo::udp::client::adopt(std::unique_ptr<boost::asio::ip::udp::socket> socket, boost::asio::ip::udp::endpoint peer, ligo::udp::client_options options) -> std::expected<ligo::udp::client, modulus::error> {
  auto state = adopt_state(std::move(socket), std::move(peer), options.reliability, options.security, options.session_id, false);
  if (!state) {
    return ligo::detail::fail<ligo::udp::client>(state.error());
  }
  return modulus::success(ligo::udp::client{
    ligo::detail::state_handle{std::move(*state)},
  });
}

auto ligo::udp::client::send(boost::json::value payload) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  return send_payload(channel_state(state_), std::move(payload));
}

auto ligo::udp::client::send(ligo::udp::message message) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  return send_payload(channel_state(state_), std::move(message.payload));
}

auto ligo::udp::client::receive() -> boost::asio::awaitable<std::expected<ligo::udp::message, modulus::error>> {
  return receive_message(channel_state(state_));
}

auto ligo::udp::client::local_endpoint() const -> std::expected<boost::asio::ip::udp::endpoint, modulus::error> {
  return local_endpoint_for(channel_state(state_));
}

auto ligo::udp::client::close() -> void {
  close_channel(channel_state(state_));
}

auto ligo::udp::client::async_close(std::chrono::milliseconds timeout) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  co_return co_await async_close_channel(channel_state(state_), timeout);
}

ligo::udp::client::client(ligo::detail::state_handle state) : state_{std::move(state)} {}

ligo::udp::server::~server() {
  close();
}

auto ligo::udp::server::operator=(ligo::udp::server&& other) noexcept -> ligo::udp::server& {
  if (this != &other) {
    close();
    state_ = std::move(other.state_);
  }
  return *this;
}

auto ligo::udp::server::bind(boost::asio::any_io_executor executor, ligo::udp::endpoint endpoint, ligo::udp::server_options options) -> std::expected<ligo::udp::server, modulus::error> {
  auto state = open_state(std::move(executor), std::move(endpoint.local), std::move(endpoint.peer), options.reliability, options.security, options.session_id, true);
  if (!state) {
    return ligo::detail::fail<ligo::udp::server>(state.error());
  }
  return modulus::success(ligo::udp::server{
    ligo::detail::state_handle{std::move(*state)},
  });
}

auto ligo::udp::server::bind(boost::asio::io_context& io, ligo::udp::endpoint endpoint, ligo::udp::server_options options) -> std::expected<ligo::udp::server, modulus::error> {
  return ligo::udp::server::bind(io.get_executor(), std::move(endpoint), options);
}

auto ligo::udp::server::adopt(std::unique_ptr<boost::asio::ip::udp::socket> socket, boost::asio::ip::udp::endpoint peer, ligo::udp::server_options options) -> std::expected<ligo::udp::server, modulus::error> {
  auto state = adopt_state(std::move(socket), std::move(peer), options.reliability, options.security, options.session_id, true);
  if (!state) {
    return ligo::detail::fail<ligo::udp::server>(state.error());
  }
  return modulus::success(ligo::udp::server{
    ligo::detail::state_handle{std::move(*state)},
  });
}

auto ligo::udp::server::send(boost::json::value payload) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  return send_payload(channel_state(state_), std::move(payload));
}

auto ligo::udp::server::send(ligo::udp::message message) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  return send_payload(channel_state(state_), std::move(message.payload));
}

auto ligo::udp::server::receive() -> boost::asio::awaitable<std::expected<ligo::udp::message, modulus::error>> {
  return receive_message(channel_state(state_));
}

auto ligo::udp::server::local_endpoint() const -> std::expected<boost::asio::ip::udp::endpoint, modulus::error> {
  return local_endpoint_for(channel_state(state_));
}

auto ligo::udp::server::close() -> void {
  close_channel(channel_state(state_));
}

auto ligo::udp::server::async_close(std::chrono::milliseconds timeout) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  co_return co_await async_close_channel(channel_state(state_), timeout);
}

ligo::udp::server::server(ligo::detail::state_handle state) : state_{std::move(state)} {}

auto ligo::udp::client_endpoint_from_config(ligo::endpoint_config const& config) -> std::expected<ligo::udp::endpoint, modulus::error> {
  return endpoint_from_config(config);
}

auto ligo::udp::server_endpoint_from_config(ligo::endpoint_config const& config) -> std::expected<ligo::udp::endpoint, modulus::error> {
  return endpoint_from_config(config);
}

auto ligo::udp::client_options_from_config(ligo::endpoint_config const& config) -> ligo::udp::client_options {
  return {
    .reliability = config.udp,
    .security = config.security,
  };
}

auto ligo::udp::server_options_from_config(ligo::endpoint_config const& config) -> ligo::udp::server_options {
  return {
    .reliability = config.udp,
    .security = config.security,
  };
}