code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#include "test-utils.h"
#include "access.h"

#include "utils/tls.h"
#include "utils/format.hpp"
#include "model/cluster.h"
#include "model/messages.h"
#include "net/names.h"
#include "net/initiator_actor.h"
#include "net/resolver_actor.h"
#include "proto/relay_support.h"
#include "transport/stream.h"
#include <rotor/asio.hpp>

using namespace syncspirit;
using namespace syncspirit::test;
using namespace syncspirit::model;
using namespace syncspirit::net;

namespace asio = boost::asio;
namespace sys = boost::system;
namespace r = rotor;
namespace ra = r::asio;

using configure_callback_t = std::function<void(r::plugin::plugin_base_t &)>;
using finish_callback_t = std::function<void()>;

auto timeout = r::pt::time_duration{r::pt::millisec{2000}};
auto host = "127.0.0.1";

struct supervisor_t : ra::supervisor_asio_t {
    using ra::supervisor_asio_t::supervisor_asio_t;

    void configure(r::plugin::plugin_base_t &plugin) noexcept override {
        ra::supervisor_asio_t::configure(plugin);
        plugin.with_casted<r::plugin::registry_plugin_t>(
            [&](auto &p) { p.register_name(names::coordinator, get_address()); });
        if (configure_callback) {
            configure_callback(plugin);
        }
    }

    void shutdown_finish() noexcept override {
        ra::supervisor_asio_t::shutdown_finish();
        if (finish_callback) {
            finish_callback();
        }
    }

    auto get_state() noexcept { return state; }

    finish_callback_t finish_callback;
    configure_callback_t configure_callback;
};

using supervisor_ptr_t = r::intrusive_ptr_t<supervisor_t>;
using actor_ptr_t = r::intrusive_ptr_t<initiator_actor_t>;

struct fixture_t {
    using acceptor_t = asio::ip::tcp::acceptor;
    using ready_ptr_t = r::intrusive_ptr_t<net::message::peer_connected_t>;
    using diff_ptr_t = r::intrusive_ptr_t<model::message::model_update_t>;
    using diff_msgs_t = std::vector<diff_ptr_t>;

    fixture_t() noexcept : ctx(io_ctx), acceptor(io_ctx), peer_sock(io_ctx) {
        test::init_logging();
        log = utils::get_logger("fixture");
    }

    virtual void finish() {
        acceptor.cancel();
        if (peer_trans) {
            peer_trans->cancel();
        }
    }

    void run() noexcept {
        auto strand = std::make_shared<asio::io_context::strand>(io_ctx);
        sup = ctx.create_supervisor<supervisor_t>().strand(strand).timeout(timeout).create_registry().finish();
        sup->configure_callback = [&](r::plugin::plugin_base_t &plugin) {
            plugin.template with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
                using connected_t = typename ready_ptr_t::element_type;
                using diff_t = typename diff_ptr_t::element_type;
                p.subscribe_actor(r::lambda<connected_t>([&](connected_t &msg) {
                    connected_message = &msg;
                    LOG_INFO(log, "received message::peer_connected_t");
                }));
                p.subscribe_actor(r::lambda<diff_t>([&](diff_t &msg) {
                    diff_msgs.emplace_back(&msg);
                    LOG_INFO(log, "received diff message");
                }));
            });
        };
        sup->finish_callback = [&]() { finish(); };
        sup->start();

        sup->create_actor<resolver_actor_t>().resolve_timeout(timeout / 2).timeout(timeout).finish();
        sup->do_process();

        my_keys = utils::generate_pair("me").value();
        peer_keys = utils::generate_pair("peer").value();

        auto md = model::device_id_t::from_cert(my_keys.cert_data).value();
        auto pd = model::device_id_t::from_cert(peer_keys.cert_data).value();

        my_device = device_t::create(md, "my-device").value();
        peer_device = device_t::create(pd, "peer-device").value();

        auto ep = asio::ip::tcp::endpoint(asio::ip::make_address(host), 0);
        acceptor.open(ep.protocol());
        acceptor.bind(ep);
        acceptor.listen();
        listening_ep = acceptor.local_endpoint();
        peer_uri = utils::parse(get_uri(listening_ep));
        log->debug("listening on {}", peer_uri);
        initiate_accept();

        cluster = new cluster_t(my_device, 1);

        cluster->get_devices().put(my_device);
        cluster->get_devices().put(peer_device);

        main();
    }

    virtual void initiate_accept() noexcept {
        acceptor.async_accept(peer_sock, [this](auto ec) { this->accept(ec); });
    }

    virtual std::string get_uri(const asio::ip::tcp::endpoint &) noexcept {
        return fmt::format("tcp://{}", listening_ep);
    }

    virtual void accept(const sys::error_code &ec) noexcept {
        LOG_INFO(log, "accept, ec: {}", ec.message());
        peer_trans = transport::initiate_tls_passive(*sup, peer_keys, std::move(peer_sock));
        initiate_peer_handshake();
    }

    virtual void initiate_peer_handshake() noexcept {
        transport::handshake_fn_t handshake_fn = [this](bool valid_peer, utils::x509_t &, const tcp::endpoint &,
                                                        const model::device_id_t *) {
            valid_handshake = valid_peer;
            on_peer_handshake();
        };
        transport::error_fn_t on_error = [](const auto &) {};
        peer_trans->async_handshake(handshake_fn, on_error);
    }

    virtual void on_peer_handshake() noexcept { LOG_INFO(log, "peer handshake"); }

    void initiate_active() noexcept {
        auto ip = asio::ip::make_address(host);
        auto ep = tcp::endpoint(ip, listening_ep.port());
        auto addresses = std::vector<tcp::endpoint>{ep};
        auto addresses_ptr = std::make_shared<decltype(addresses)>(addresses);
        peer_trans = transport::initiate_tls_active(*sup, peer_keys, my_device->device_id(), peer_uri);

        transport::error_fn_t on_error = [&](auto &ec) {
            LOG_WARN(log, "initiate_active/connect, err: {}", ec.message());
        };
        transport::connect_fn_t on_connect = [&](auto) {
            LOG_INFO(log, "initiate_active/peer connect");
            active_connect();
        };

        peer_trans->async_connect(addresses_ptr, on_connect, on_error);
    }

    virtual void active_connect() {
        LOG_TRACE(log, "active_connect");
        transport::handshake_fn_t handshake_fn = [this](bool, utils::x509_t &, const tcp::endpoint &,
                                                        const model::device_id_t *) {
            valid_handshake = true;
            LOG_INFO(log, "test_passive_success/peer handshake");
        };
        transport::error_fn_t on_hs_error = [&](const auto &ec) {
            LOG_WARN(log, "test_passive_success/peer handshake, err: {}", ec.message());
        };
        peer_trans->async_handshake(handshake_fn, on_hs_error);
    }

    virtual void main() noexcept {}

    virtual actor_ptr_t create_actor() noexcept {
        return sup->create_actor<initiator_actor_t>()
            .timeout(timeout)
            .peer_device_id(peer_device->device_id())
            .relay_session(relay_session)
            .relay_enabled(true)
            .uris(utils::uri_container_t{peer_uri})
            .cluster(use_model ? cluster : nullptr)
            .sink(sup->get_address())
            .ssl_pair(&my_keys)
            .router(*sup)
            .escalate_failure()
            .finish();
    }

    virtual actor_ptr_t create_passive_actor() noexcept {
        return sup->create_actor<initiator_actor_t>()
            .timeout(timeout)
            .sock(std::move(peer_sock))
            .ssl_pair(&my_keys)
            .router(*sup)
            .cluster(cluster)
            .sink(sup->get_address())
            .escalate_failure()
            .finish();
    }

    cluster_ptr_t cluster;
    asio::io_context io_ctx{1};
    ra::system_context_asio_t ctx;
    acceptor_t acceptor;
    supervisor_ptr_t sup;
    asio::ip::tcp::endpoint listening_ep;
    utils::logger_t log;
    asio::ip::tcp::socket peer_sock;
    config::bep_config_t bep_config;
    utils::key_pair_t my_keys;
    utils::key_pair_t peer_keys;
    utils::uri_ptr_t peer_uri;
    model::device_ptr_t my_device;
    model::device_ptr_t peer_device;
    transport::stream_sp_t peer_trans;
    ready_ptr_t connected_message;
    diff_msgs_t diff_msgs;
    utils::bytes_t relay_session;
    bool use_model = true;

    bool valid_handshake = false;
};

void test_connect_timeout() {
    struct F : fixture_t {
        void initiate_accept() noexcept override {}
        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
        }
    };
    F().run();
}

void test_connect_unsupported_proto() {
    struct F : fixture_t {
        std::string get_uri(const asio::ip::tcp::endpoint &) noexcept override {
            return fmt::format("xxx://{}", listening_ep);
        }
        void main() noexcept override {
            create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
        }
    };
    F().run();
}

void test_handshake_timeout() {
    struct F : fixture_t {

        void accept(const sys::error_code &ec) noexcept override { LOG_INFO(log, "accept (ignoring)", ec.message()); }

        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
            REQUIRE(diff_msgs.size() == 2);
            CHECK(diff_msgs[0]->payload.diff->apply(*cluster, get_apply_controller()));
            CHECK(peer_device->get_state() == device_state_t::connecting);
            CHECK(diff_msgs[1]->payload.diff->apply(*cluster, get_apply_controller()));
            CHECK(peer_device->get_state() == device_state_t::offline);
        }
    };
    F().run();
}

void test_handshake_garbage() {
    struct F : fixture_t {

        void accept(const sys::error_code &) noexcept override {
            auto buff = asio::buffer("garbage-garbage-garbage");
            peer_sock.write_some(buff);
        }

        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
            REQUIRE(diff_msgs.size() == 2);
            CHECK(diff_msgs[0]->payload.diff->apply(*cluster, get_apply_controller()));
            CHECK(peer_device->get_state() == device_state_t::connecting);
            CHECK(diff_msgs[1]->payload.diff->apply(*cluster, get_apply_controller()));
            CHECK(peer_device->get_state() == device_state_t::offline);
        }
    };
    F().run();
}

void test_connection_refused() {
    struct F : fixture_t {

        std::string get_uri(const asio::ip::tcp::endpoint &) noexcept override {
            return fmt::format("tcp://{}:0", host);
        }

        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
        }
    };
    F().run();
}

void test_connection_refused_no_model() {
    struct F : fixture_t {
        F() { use_model = false; }

        std::string get_uri(const asio::ip::tcp::endpoint &) noexcept override {
            return fmt::format("tcp://{}:0", host);
        }

        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
        }
    };
    F().run();
}

void test_resolve_failure() {
    struct F : fixture_t {

        std::string get_uri(const asio::ip::tcp::endpoint &) noexcept override { return "tcp://x.example.com"; }

        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
        }
    };
    F().run();
}

void test_success() {
    struct F : fixture_t {
        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::OPERATIONAL);
            REQUIRE(connected_message);
            CHECK(connected_message->payload.proto == "tcp");
            CHECK(connected_message->payload.peer_device_id == peer_device->device_id());
            CHECK(valid_handshake);
            sup->do_shutdown();
            sup->do_process();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            REQUIRE(diff_msgs.size() == 1);
            CHECK(diff_msgs[0]->payload.diff->apply(*cluster, get_apply_controller()));
            CHECK(peer_device->get_state() == device_state_t::connecting);
        }
    };
    F().run();
}

void test_success_no_model() {
    struct F : fixture_t {
        F() { use_model = false; }

        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::OPERATIONAL);
            CHECK(connected_message);
            CHECK(connected_message->payload.peer_device_id == peer_device->device_id());
            CHECK(valid_handshake);
            sup->do_shutdown();
            sup->do_process();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            REQUIRE(diff_msgs.size() == 0);
        }
    };
    F().run();
}

struct passive_fixture_t : fixture_t {
    actor_ptr_t act;
    bool active_connect_invoked = false;

    void active_connect() override {
        LOG_TRACE(log, "active_connect");
        if (!act || active_connect_invoked) {
            return;
        }
        active_connect_invoked = true;
        active_connect_impl();
    }

    virtual void active_connect_impl() { fixture_t::active_connect(); }

    void accept(const sys::error_code &ec) noexcept override {
        LOG_INFO(log, "test_passive_success/accept, ec: {}", ec.message());
        act = create_passive_actor();
        sup->do_process();
        active_connect();
    }
};

void test_passive_success() {
    struct F : passive_fixture_t {
        void main() noexcept override {
            initiate_active();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::OPERATIONAL);
            REQUIRE(connected_message);
            CHECK(connected_message->payload.proto == "tcp");
            CHECK(connected_message->payload.peer_device_id == peer_device->device_id());
            CHECK(valid_handshake);
            sup->do_shutdown();
            sup->do_process();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
        }
    };
    F().run();
}

void test_passive_garbage() {
    struct F : passive_fixture_t {

        tcp::socket client_sock;
        tcp::resolver::results_type addresses;

        F() : client_sock{io_ctx} {}

        void active_connect_impl() noexcept override {
            tcp::resolver resolver(io_ctx);
            addresses = resolver.resolve(host, std::to_string(listening_ep.port()));
            asio::async_connect(client_sock, addresses.begin(), addresses.end(), [&](auto ec, auto) {
                LOG_INFO(log, "test_passive_garbage/peer connect, ec: {}", ec.message());
                auto buff = asio::buffer("garbage-garbage-garbage");
                client_sock.write_some(buff);
                sup->do_process();
            });
        }

        void main() noexcept override {
            initiate_active();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
        }
    };
    F().run();
}

void test_passive_timeout() {
    struct F : passive_fixture_t {

        void active_connect() noexcept override { LOG_INFO(log, "test_passive_timeout/active_connect NOOP"); }

        void main() noexcept override {
            initiate_active();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
        }
    };
    F().run();
}

struct passive_relay_fixture_t : fixture_t {
    utils::bytes_t rx_buff;
    bool initiate_handshake = true;
    passive_relay_fixture_t() {
        relay_session = as_owned_bytes("relay-session-key");
        rx_buff.resize(128);
    }

    void on_read(size_t bytes) noexcept {
        LOG_TRACE(log, "read (relay/passive), {} bytes", bytes);
        auto r = proto::relay::parse({rx_buff.data(), bytes});
        auto &wrapped = std::get<proto::relay::wrapped_message_t>(r);
        auto &msg = std::get<proto::relay::join_session_request_t>(wrapped.message);
        CHECK(msg.key == relay_session);
        relay_reply();
    }

    virtual void on_write(size_t bytes) noexcept {
        LOG_TRACE(log, "write (relay/passive), {} bytes", bytes);

        if (initiate_handshake) {
            auto upgradeable = static_cast<transport::upgradeable_stream_base_t *>(peer_trans.get());
            auto ssl = transport::ssl_junction_t{my_device->device_id(), &peer_keys, false, "bep"};
            peer_trans = upgradeable->upgrade(ssl, true);
            initiate_peer_handshake();
        }
    }

    virtual void relay_reply() noexcept { write(proto::relay::response_t{0, "success"}); }

    virtual void write(const proto::relay::message_t &msg) noexcept {
        proto::relay::serialize(msg, rx_buff);
        transport::error_fn_t err_fn([&](auto ec) { log->error("(relay/passive), read_err: {}", ec.message()); });
        transport::io_fn_t write_fn = [this](size_t bytes) { on_write(bytes); };
        peer_trans->async_send(asio::buffer(rx_buff), write_fn, err_fn);
    }

    void accept(const sys::error_code &ec) noexcept override {
        LOG_INFO(log, "accept (relay/passive), ec: {}", ec.message());
        auto uri = utils::parse("tcp://127.0.0.1:0/");
        auto cfg = transport::transport_config_t{{}, uri, *sup, std::move(peer_sock), false};
        peer_trans = transport::initiate_stream(cfg);

        transport::error_fn_t read_err_fn([&](auto ec) { log->error("(relay/passive), read_err: {}", ec.message()); });
        transport::io_fn_t read_fn = [this](size_t bytes) { on_read(bytes); };
        peer_trans->async_recv(asio::buffer(rx_buff), read_fn, read_err_fn);
    }
};

void test_relay_passive_success() {
    struct F : passive_relay_fixture_t {
        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::OPERATIONAL);
            REQUIRE(connected_message);
            CHECK(connected_message->payload.proto == "relay");
            CHECK(connected_message->payload.peer_device_id == peer_device->device_id());
            CHECK(valid_handshake);
            sup->do_shutdown();
            sup->do_process();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(diff_msgs.size() == 0);
        }
    };
    F().run();
}

void test_relay_passive_garbage() {
    struct F : passive_relay_fixture_t {

        void write(const proto::relay::message_t &) noexcept override {
            rx_buff = as_owned_bytes("garbage-garbage-garbage");
            initiate_handshake = false;
            transport::error_fn_t err_fn([&](auto ec) { log->error("(relay/passive), read_err: {}", ec.message()); });
            transport::io_fn_t write_fn = [this](size_t bytes) { on_write(bytes); };
            peer_trans->async_send(asio::buffer(rx_buff), write_fn, err_fn);
        }

        void main() noexcept override {
            create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
            CHECK(!valid_handshake);
            sup->do_shutdown();
            sup->do_process();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(diff_msgs.size() == 0);
        }
    };
    F().run();
}

void test_relay_passive_wrong_message() {
    struct F : passive_relay_fixture_t {

        void relay_reply() noexcept override { write(proto::relay::pong_t{}); }

        void main() noexcept override {
            initiate_handshake = false;
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
            CHECK(!valid_handshake);
            sup->do_shutdown();
            sup->do_process();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(diff_msgs.size() == 0);
        }
    };
    F().run();
}

void test_relay_passive_unsuccessful_join() {
    struct F : passive_relay_fixture_t {

        void relay_reply() noexcept override { write(proto::relay::response_t{5, "some-fail-reason"}); }

        void main() noexcept override {
            initiate_handshake = false;
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
            CHECK(!valid_handshake);
            sup->do_shutdown();
            sup->do_process();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(diff_msgs.size() == 0);
        }
    };
    F().run();
}

void test_relay_malformed_uri() {
    struct F : fixture_t {
        std::string get_uri(const asio::ip::tcp::endpoint &) noexcept override {
            return fmt::format("relay://{}", listening_ep);
        }

        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
            CHECK(!valid_handshake);
            sup->do_shutdown();
            sup->do_process();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(diff_msgs.size() == 2);
        }
    };
    F().run();
}

void test_relay_active_wrong_relay_device_id() {
    struct F : fixture_t {

        std::string get_uri(const asio::ip::tcp::endpoint &) noexcept override {
            return fmt::format("relay://{}?id={}", listening_ep, my_device->device_id().get_value());
        }

        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
            CHECK(!valid_handshake);
            sup->do_shutdown();
            sup->do_process();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(diff_msgs.size() == 2);
        }
    };
    F().run();
}

struct active_relay_fixture_t : fixture_t {
    utils::key_pair_t relay_keys;
    model::device_id_t relay_device;
    utils::bytes_t rx_buff;
    utils::bytes_t session_key = as_owned_bytes("lorem-session-dolor");
    transport::stream_sp_t relay_trans;
    bool session_mode = false;

    active_relay_fixture_t() {
        relay_keys = utils::generate_pair("relay").value();
        relay_device = model::device_id_t::from_cert(relay_keys.cert_data).value();
        rx_buff.resize(128);
    }

    std::string get_uri(const asio::ip::tcp::endpoint &) noexcept override {
        return fmt::format("relay://{}?id={}", listening_ep, relay_device.get_value());
    }

    void accept(const sys::error_code &ec) noexcept override {
        LOG_INFO(log, "relay/accept, ec: {}", ec.message());
        if (!session_mode) {
            relay_trans = transport::initiate_tls_passive(*sup, relay_keys, std::move(peer_sock));
            transport::handshake_fn_t handshake_fn = [this](bool valid_peer, utils::x509_t &, const tcp::endpoint &,
                                                            const model::device_id_t *) {
                valid_handshake = valid_peer;
                on_relay_handshake();
            };
            transport::error_fn_t on_error = [](const auto &) {};
            relay_trans->async_handshake(handshake_fn, on_error);
            return;
        }
        auto uri = utils::parse("tcp://127.0.0.1:0/");
        auto cfg = transport::transport_config_t{{}, uri, *sup, std::move(peer_sock), false};
        peer_trans = transport::initiate_stream(cfg);

        transport::error_fn_t read_err_fn([&](auto ec) { log->error("(relay/active), read_err: {}", ec.message()); });
        transport::io_fn_t read_fn = [this](size_t bytes) { on_read_peer(bytes); };
        peer_trans->async_recv(asio::buffer(rx_buff), read_fn, read_err_fn);
    }

    virtual void on_relay_handshake() noexcept {
        transport::error_fn_t read_err_fn([&](auto ec) { log->error("(relay/active), read_err: {}", ec.message()); });
        transport::io_fn_t read_fn = [this](size_t bytes) { on_read(bytes); };
        relay_trans->async_recv(asio::buffer(rx_buff), read_fn, read_err_fn);
    }

    virtual void relay_reply() noexcept {
        auto sha256 = peer_device->device_id().get_sha256();
        auto id = utils::bytes_t(sha256.begin(), sha256.end());
        write(relay_trans, proto::relay::session_invitation_t{std::move(id), session_key, asio::ip::address_v4(0),
                                                              listening_ep.port(), false});
    }

    virtual void session_reply() noexcept { write(peer_trans, proto::relay::response_t{0, "ok"}); }

    virtual void write(transport::stream_sp_t &stream, const proto::relay::message_t &msg) noexcept {
        proto::relay::serialize(msg, rx_buff);
        transport::error_fn_t err_fn([&](auto ec) { log->error("(relay/passive), read_err: {}", ec.message()); });
        transport::io_fn_t write_fn = [this](size_t bytes) { on_write(bytes); };
        stream->async_send(asio::buffer(rx_buff), write_fn, err_fn);
    }

    virtual void on_read_peer(size_t bytes) {
        log->debug("(relay/active) read peer {} bytes", bytes);
        auto r = proto::relay::parse({rx_buff.data(), bytes});
        auto &wrapped = std::get<proto::relay::wrapped_message_t>(r);
        auto &msg = std::get<proto::relay::join_session_request_t>(wrapped.message);
        CHECK(msg.key == session_key);
        session_reply();
    }

    virtual void on_read(size_t bytes) {
        log->debug("(relay/active) read {} bytes", bytes);
        auto r = proto::relay::parse({rx_buff.data(), bytes});
        auto &wrapped = std::get<proto::relay::wrapped_message_t>(r);
        auto &msg = std::get<proto::relay::connect_request_t>(wrapped.message);
        CHECK(msg.device_id == peer_device->device_id().get_sha256());
        relay_reply();
    }

    virtual void on_write(size_t bytes) {
        log->debug("(relay/active) write {} bytes", bytes);
        if (!session_mode) {
            acceptor.async_accept(peer_sock, [this](auto ec) { this->accept(ec); });
            session_mode = true;
        } else {
            auto upgradeable = static_cast<transport::upgradeable_stream_base_t *>(peer_trans.get());
            auto ssl = transport::ssl_junction_t{my_device->device_id(), &peer_keys, false, "bep"};
            peer_trans = upgradeable->upgrade(ssl, false);
            initiate_peer_handshake();
        }
    }
};

void test_relay_active_success() {
    struct F : active_relay_fixture_t {
        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::OPERATIONAL);
            REQUIRE(connected_message);
            CHECK(connected_message->payload.proto == "relay");
            CHECK(connected_message->payload.peer_device_id == peer_device->device_id());
            CHECK(valid_handshake);
            sup->do_shutdown();
            sup->do_process();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            REQUIRE(diff_msgs.size() == 1);
            CHECK(diff_msgs[0]->payload.diff->apply(*cluster, get_apply_controller()));
            CHECK(peer_device->get_state() == device_state_t::connecting);
        }
    };
    F().run();
}

void test_relay_active_not_enabled() {
    struct F : active_relay_fixture_t {

        actor_ptr_t create_actor() noexcept override {
            return sup->create_actor<initiator_actor_t>()
                .timeout(timeout)
                .peer_device_id(peer_device->device_id())
                .relay_session(relay_session)
                .uris({peer_uri})
                .cluster(use_model ? cluster : nullptr)
                .sink(sup->get_address())
                .ssl_pair(&my_keys)
                .router(*sup)
                .escalate_failure()
                .finish();
        }

        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            sup->do_shutdown();
            sup->do_process();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(peer_device->get_state() == device_state_t::offline);
        }
    };
    F().run();
}

void test_relay_wrong_device() {
    struct F : active_relay_fixture_t {

        void relay_reply() noexcept override {
            auto sha256 = relay_device.get_sha256();
            auto id = utils::bytes_t(sha256.begin(), sha256.end());
            write(relay_trans, proto::relay::session_invitation_t{std::move(id), session_key, asio::ip::address_v4(0),
                                                                  listening_ep.port(), false});
        }
        void on_write(size_t) override {}

        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
            CHECK(valid_handshake);
            sup->do_shutdown();
            sup->do_process();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(diff_msgs.size() == 2);
        }
    };
    F().run();
}

void test_relay_non_connectable() {
    struct F : active_relay_fixture_t {

        void relay_reply() noexcept override {
            auto sha256 = peer_device->device_id().get_sha256();
            auto id = utils::bytes_t(sha256.begin(), sha256.end());
            write(relay_trans, proto::relay::session_invitation_t{std::move(id), session_key,
                                                                  boost::asio::ip::address_v4(0), 0, false});
        }

        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
            sup->do_shutdown();
            sup->do_process();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(diff_msgs.size() == 2);
        }
    };
    F().run();
}

void test_relay_garbage_reply() {
    struct F : active_relay_fixture_t {

        void write(transport::stream_sp_t &stream, const proto::relay::message_t &) noexcept override {
            rx_buff = as_owned_bytes("garbage-garbage-garbage");
            transport::error_fn_t err_fn([&](auto ec) { log->error("(relay/passive), read_err: {}", ec.message()); });
            transport::io_fn_t write_fn = [this](size_t bytes) { on_write(bytes); };
            stream->async_send(asio::buffer(rx_buff), write_fn, err_fn);
        }

        void on_write(size_t) override {}

        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
            sup->do_shutdown();
            sup->do_process();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(diff_msgs.size() == 2);
        }
    };
    F().run();
}

void test_relay_non_invitation_reply() {
    struct F : active_relay_fixture_t {

        void relay_reply() noexcept override { write(relay_trans, proto::relay::pong_t{}); }
        void on_write(size_t) override {}

        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(!connected_message);
            sup->do_shutdown();
            sup->do_process();
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
            CHECK(diff_msgs.size() == 2);
        }
    };
    F().run();
}

int _init() {
    REGISTER_TEST_CASE(test_connect_unsupported_proto, "test_connect_unsupported_proto", "[initiator]");
    REGISTER_TEST_CASE(test_connect_timeout, "test_connect_timeout", "[initiator]");
    REGISTER_TEST_CASE(test_handshake_timeout, "test_handshake_timeout", "[initiator]");
    REGISTER_TEST_CASE(test_handshake_garbage, "test_handshake_garbage", "[initiator]");
    REGISTER_TEST_CASE(test_connection_refused, "test_connection_refused", "[initiator]");
    REGISTER_TEST_CASE(test_connection_refused_no_model, "test_connection_refused_no_model", "[initiator]");
    REGISTER_TEST_CASE(test_resolve_failure, "test_resolve_failure", "[initiator]");
    REGISTER_TEST_CASE(test_success, "test_success", "[initiator]");
    REGISTER_TEST_CASE(test_success_no_model, "test_success_no_model", "[initiator]");
    REGISTER_TEST_CASE(test_passive_success, "test_passive_success", "[initiator]");
    REGISTER_TEST_CASE(test_passive_garbage, "test_passive_garbage", "[initiator]");
    REGISTER_TEST_CASE(test_passive_timeout, "test_passive_timeout", "[initiator]");
    REGISTER_TEST_CASE(test_relay_passive_success, "test_relay_passive_success", "[initiator]");
    REGISTER_TEST_CASE(test_relay_passive_garbage, "test_relay_passive_garbage", "[initiator]");
    REGISTER_TEST_CASE(test_relay_passive_wrong_message, "test_relay_passive_wrong_message", "[initiator]");
    REGISTER_TEST_CASE(test_relay_passive_unsuccessful_join, "test_relay_passive_unsuccessful_join", "[initiator]");
    REGISTER_TEST_CASE(test_relay_malformed_uri, "test_relay_malformed_uri", "[initiator]");
    REGISTER_TEST_CASE(test_relay_active_wrong_relay_device_id, "test_relay_active_wrong_relay_device_id",
                       "[initiator]");
    REGISTER_TEST_CASE(test_relay_active_success, "test_relay_active_success", "[initiator]");
    REGISTER_TEST_CASE(test_relay_active_not_enabled, "test_relay_active_not_enabled", "[initiator]");
    REGISTER_TEST_CASE(test_relay_wrong_device, "test_relay_wrong_device", "[initiator]");
    REGISTER_TEST_CASE(test_relay_non_connectable, "test_relay_non_connectable", "[initiator]");
    REGISTER_TEST_CASE(test_relay_garbage_reply, "test_relay_garbage_reply", "[initiator]");
    REGISTER_TEST_CASE(test_relay_non_invitation_reply, "test_relay_non_invitation_reply", "[initiator]");

    return 1;
}

static int v = _init();


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#include "test-utils.h"
#include "access.h"

#include "utils/tls.h"
#include "utils/format.hpp"
#include "model/cluster.h"
#include "model/messages.h"
#include "model/diff/cluster_visitor.h"
#include "model/diff/contact/relay_connect_request.h"
#include "model/diff/contact/unknown_connected.h"
#include "net/names.h"
#include "net/messages.h"
#include "net/relay_actor.h"
#include "transport/stream.h"
#include <rotor/asio.hpp>
#include <boost/algorithm/string/replace.hpp>

using namespace syncspirit;
using namespace syncspirit::test;
using namespace syncspirit::model;
using namespace syncspirit::net;

namespace asio = boost::asio;
namespace sys = boost::system;
namespace r = rotor;
namespace ra = r::asio;

using configure_callback_t = std::function<void(r::plugin::plugin_base_t &)>;

auto timeout = r::pt::time_duration{r::pt::millisec{1500}};
auto host = "127.0.0.1";

struct supervisor_t : ra::supervisor_asio_t {
    using parent_t = ra::supervisor_asio_t;
    using parent_t::parent_t;

    void configure(r::plugin::plugin_base_t &plugin) noexcept override {
        parent_t::configure(plugin);
        plugin.with_casted<r::plugin::registry_plugin_t>([&](auto &p) {
            p.register_name(names::coordinator, get_address());
            p.register_name(names::peer_supervisor, get_address());
            p.register_name(names::http11_relay, get_address());
        });
        if (configure_callback) {
            configure_callback(plugin);
        }
    }

    void on_child_shutdown(actor_base_t *actor) noexcept override {
        if (actor) {
            spdlog::info("child shutdown: {}, reason: {}", actor->get_identity(),
                         actor->get_shutdown_reason()->message());
        }
        parent_t::on_child_shutdown(actor);
    }

    void shutdown_finish() noexcept override {
        parent_t::shutdown_finish();
        if (acceptor) {
            acceptor->cancel();
        }
    }

    auto get_state() noexcept { return state; }

    asio::ip::tcp::acceptor *acceptor = nullptr;
    configure_callback_t configure_callback;
};

using supervisor_ptr_t = r::intrusive_ptr_t<supervisor_t>;
using actor_ptr_t = r::intrusive_ptr_t<relay_actor_t>;

struct fixture_t : private model::diff::cluster_visitor_t {
    using acceptor_t = asio::ip::tcp::acceptor;

    fixture_t() noexcept : ctx(io_ctx), acceptor(io_ctx), peer_sock(io_ctx) {
        test::init_logging();
        log = utils::get_logger("fixture");
        relay_config = config::relay_config_t{
            true,
            true,
            utils::parse("https://some-endpoint.com/"),
            1024 * 1024,
        };
    }

    void run() noexcept {
        auto strand = std::make_shared<asio::io_context::strand>(io_ctx);
        sup = ctx.create_supervisor<supervisor_t>().strand(strand).timeout(timeout).create_registry().finish();
        sup->configure_callback = [&](r::plugin::plugin_base_t &plugin) {
            plugin.template with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
                using model_update_t = model::message::model_update_t;
                p.subscribe_actor(r::lambda<model_update_t>([&](model_update_t &msg) { on(msg); }));
                using http_req_t = net::message::http_request_t;
                p.subscribe_actor(r::lambda<http_req_t>([&](http_req_t &req) {
                    LOG_INFO(log, "received http request");
                    http::response<http::string_body> res;
                    res.result(200);
                    res.body() = public_relays;
                    sup->reply_to(req, std::move(res), public_relays.size());
                }));
                using connect_req_t = net::message::connect_request_t;
                p.subscribe_actor(r::lambda<connect_req_t>([&](connect_req_t &req) {
                    LOG_INFO(log, "(connect request)");
                    on(req);
                }));
            });
        };
        sup->start();
        sup->do_process();

        auto ep = asio::ip::tcp::endpoint(asio::ip::make_address(host), 0);
        acceptor.open(ep.protocol());
        acceptor.bind(ep);
        acceptor.listen();
        listening_ep = acceptor.local_endpoint();

        my_keys = utils::generate_pair("me").value();
        relay_keys = utils::generate_pair("relay").value();
        peer_keys = utils::generate_pair("peer").value();

        auto md = model::device_id_t::from_cert(my_keys.cert_data).value();
        auto rd = model::device_id_t::from_cert(relay_keys.cert_data).value();
        auto pd = model::device_id_t::from_cert(peer_keys.cert_data).value();

        my_device = device_t::create(md, "my-device").value();
        relay_device = device_t::create(rd, "relay-device").value();
        peer_device = device_t::create(rd, "peer-device").value();

        public_relays = generate_public_relays(listening_ep, relay_device);
        log->debug("public relays json: {}", public_relays);
        initiate_accept();

        cluster = new cluster_t(my_device, 1);

        cluster->get_devices().put(my_device);
        cluster->get_devices().put(peer_device);

        session_key = as_owned_bytes("lorem-imspum-dolor");

        main();
    }

    virtual void main() noexcept {}

    virtual std::string generate_public_relays(const asio::ip::tcp::endpoint &,
                                               model::device_ptr_t &relay_device) noexcept {
        std::string pattern = R""(
    {
      "relays": [
        {
          "url": "##URL##&pingInterval=0m1s&networkTimeout=2m0s&sessionLimitBps=0&globalLimitBps=0&statusAddr=:22070&providedBy=ina",
          "location": {
            "latitude": 50.1049,
            "longitude": 8.6295,
            "city": "Frankfurt am Main",
            "country": "DE",
            "continent": "EU"
          }
        }
    ]
    }
        )"";
        auto url = fmt::format("relay://{}/?id={}", listening_ep, relay_device->device_id().get_value());
        return boost::algorithm::replace_first_copy(pattern, "##URL##", url);
    }

    virtual void initiate_accept() noexcept {
        acceptor.async_accept(peer_sock, [this](auto ec) { this->accept(ec); });
        sup->acceptor = &acceptor;
    }

    virtual void accept(const sys::error_code &ec) noexcept {
        LOG_INFO(log, "accept (relay), ec: {}, remote = {}", ec.message(), peer_sock.remote_endpoint());
        auto uri = utils::parse("tcp://127.0.0.1:0/");
        auto cfg = transport::transport_config_t{{}, uri, *sup, std::move(peer_sock), false};
        relay_trans = transport::initiate_stream(cfg);
        relay_read();
    }

    virtual actor_ptr_t create_actor() noexcept {
        return sup->create_actor<actor_ptr_t::element_type>()
            .timeout(timeout)
            .cluster(cluster)
            .relay_config(relay_config)
            .escalate_failure()
            .finish();
    }

    virtual void on(net::message::connect_request_t &req) noexcept {
        auto &uri = req.payload.request_payload.uri;
        log->info("requested connect to {}", uri);
        auto cfg = transport::transport_config_t{{}, uri, *sup, {}, true};

        auto ip = asio::ip::make_address(host);
        auto peer_ep = tcp::endpoint(ip, uri->port_number());
        auto addresses = std::vector<tcp::endpoint>{peer_ep};
        auto addresses_ptr = std::make_shared<decltype(addresses)>(addresses);
        auto trans = transport::initiate_stream(cfg);
        transport::error_fn_t on_error = [&](auto &ec) { LOG_WARN(log, "active/connect, err: {}", ec.message()); };
        using ptr_t = model::intrusive_ptr_t<std::decay_t<decltype(req)>>;
        auto ptr = ptr_t(&req);
        transport::connect_fn_t on_connect = [ptr, trans, addresses_ptr, this](const tcp::endpoint &ep) {
            LOG_INFO(log, "active/connected");
            sup->reply_to(*ptr, trans, ep);
        };
        trans->async_connect(addresses_ptr, on_connect, on_error);
    }

    void send_relay(const proto::relay::message_t &msg) noexcept {
        proto::relay::serialize(msg, relay_tx);
        transport::error_fn_t on_error = [&](auto &ec) { LOG_WARN(log, "relay/write, err: {}", ec.message()); };
        transport::io_fn_t on_write = [&](size_t bytes) { LOG_TRACE(log, "relay/write, {} bytes", bytes); };
        relay_trans->async_send(asio::buffer(relay_tx), on_write, on_error);
    }

    virtual void on_relay(proto::relay::ping_t &) noexcept {};
    virtual void on_relay(proto::relay::pong_t &) noexcept {};
    virtual void on_relay(proto::relay::join_relay_request_t &) noexcept {
        LOG_INFO(log, "join_relay_request_t");
        send_relay(proto::relay::response_t{0, "ok"});
    };
    virtual void on_relay(proto::relay::join_session_request_t &) noexcept {};
    virtual void on_relay(proto::relay::response_t &) noexcept {};
    virtual void on_relay(proto::relay::connect_request_t &) noexcept {};
    virtual void on_relay(proto::relay::session_invitation_t &) noexcept {};
    virtual void on(model::message::model_update_t &update) noexcept {
        auto &diff = *update.payload.diff;
        auto r = diff.apply(*cluster, get_apply_controller());
        if (!r) {
            LOG_ERROR(log, "error applying diff: {}", r.error().message());
        }
        r = diff.visit(*this, nullptr);
        if (!r) {
            LOG_ERROR(log, "error visiting diff: {}", r.error().message());
        }
    }

    void relay_read() noexcept {
        transport::error_fn_t on_error = [&](auto &ec) { LOG_WARN(log, "relay/read, err: {}", ec.message()); };
        transport::io_fn_t on_read = [&](size_t bytes) {
            LOG_TRACE(log, "relay/read, {} bytes", bytes);
            auto msg = proto::relay::parse({relay_rx.data(), bytes});
            auto wrapped = std::get_if<proto::relay::wrapped_message_t>(&msg);
            if (!wrapped) {
                LOG_ERROR(log, "relay/read non-message?");
                return;
            }
            std::visit([&](auto &it) { on_relay(it); }, wrapped->message);
        };
        relay_rx.resize(1500);
        auto buff = asio::buffer(relay_rx.data(), relay_rx.size());
        relay_trans->async_recv(buff, on_read, on_error);
        LOG_TRACE(log, "relay/async recv");
    }

    config::relay_config_t relay_config;
    cluster_ptr_t cluster;
    asio::io_context io_ctx;
    ra::system_context_asio_t ctx;
    acceptor_t acceptor;
    supervisor_ptr_t sup;
    asio::ip::tcp::endpoint listening_ep;
    utils::logger_t log;
    asio::ip::tcp::socket peer_sock;
    std::string public_relays;
    utils::key_pair_t my_keys;
    utils::key_pair_t relay_keys;
    utils::key_pair_t peer_keys;
    model::device_ptr_t my_device;
    model::device_ptr_t relay_device;
    model::device_ptr_t peer_device;
    transport::stream_sp_t relay_trans;
    utils::bytes_t relay_rx;
    utils::bytes_t relay_tx;
    utils::bytes_t session_key;
};

void test_master_connect() {
    struct F : fixture_t {
        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();

            CHECK(sup->get_state() == r::state_t::OPERATIONAL);
            REQUIRE(my_device->get_uris().size() == 1);
            CHECK(my_device->get_uris()[0]->scheme() == "relay");

            sup->shutdown();
            io_ctx.restart();
            io_ctx.run();

            CHECK(my_device->get_uris().size() == 0);
            io_ctx.restart();
            io_ctx.run();

            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
        }

        void on(model::message::model_update_t &update) noexcept override {
            LOG_INFO(log, "contact_update_t");
            fixture_t::on(update);
            io_ctx.stop();
        }
    };

    F().run();
}

void test_passive() {
    struct F : fixture_t {
        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(sent);
            CHECK(received);
            CHECK(sup->get_state() == r::state_t::OPERATIONAL);

            sup->shutdown();
            io_ctx.restart();
            io_ctx.run();

            CHECK(my_device->get_uris().size() == 0);
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
        }

        void on(model::message::model_update_t &update) noexcept override {
            LOG_INFO(log, "contact_update_t");
            fixture_t::on(update);
            if (my_device->get_uris().size() == 1 && !sent) {
                sent = true;
                auto sha256 = peer_device->device_id().get_sha256();
                auto id = utils::bytes_t(sha256.begin(), sha256.end());
                auto msg = proto::relay::session_invitation_t{std::move(id), session_key, {}, 12345, true};
                send_relay(msg);
            }
        }

        outcome::result<void> operator()(const model::diff::contact::relay_connect_request_t &diff,
                                         void *) noexcept override {
            CHECK(diff.peer == peer_device->device_id());
            CHECK(diff.session_key == session_key);
            CHECK(diff.relay.port() == 12345);
            CHECK(diff.relay.address().to_string() == "127.0.0.1");
            received = true;
            io_ctx.stop();
            return outcome::success();
        }

        bool sent = false;
        bool received = false;
    };
    F().run();
}

void test_passive_ping_pong() {
    struct F : fixture_t {
        void main() noexcept override {
            auto act = create_actor();
            io_ctx.run();
            CHECK(received);
            CHECK(sup->get_state() == r::state_t::OPERATIONAL);

            sup->shutdown();
            io_ctx.restart();
            io_ctx.run();

            CHECK(my_device->get_uris().size() == 0);
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
        }

        void on_relay(proto::relay::join_relay_request_t &request) noexcept override {
            fixture_t::on_relay(request);
            relay_read();
        };

        void on_relay(proto::relay::ping_t &message) noexcept override {
            fixture_t::on_relay(message);
            send_relay(proto::relay::pong_t{});
            auto sha256 = peer_device->device_id().get_sha256();
            auto id = utils::bytes_t(sha256.begin(), sha256.end());
            auto msg = proto::relay::session_invitation_t{std::move(id), session_key, {}, 12345, true};
            send_relay(msg);
        };

        outcome::result<void> operator()(const model::diff::contact::relay_connect_request_t &diff,
                                         void *) noexcept override {
            CHECK(diff.peer == peer_device->device_id());
            CHECK(diff.session_key == session_key);
            CHECK(diff.relay.port() == 12345);
            CHECK(diff.relay.address().to_string() == "127.0.0.1");
            received = true;
            io_ctx.stop();
            return outcome::success();
        }

        bool received = false;
    };
    F().run();
}

void test_passive_unknown() {
    struct F : fixture_t {
        void main() noexcept override {
            auto act = create_actor();
            cluster->get_devices().remove(peer_device);
            io_ctx.run();
            CHECK(sent);
            CHECK(unknownon_connected);
            CHECK(sup->get_state() == r::state_t::OPERATIONAL);

            sup->shutdown();
            io_ctx.restart();
            io_ctx.run();

            CHECK(my_device->get_uris().size() == 0);
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
        }

        void on(model::message::model_update_t &update) noexcept override {
            LOG_INFO(log, "contact_update_t");
            fixture_t::on(update);
            if (my_device->get_uris().size() == 1 && !sent) {
                sent = true;
                auto sha256 = peer_device->device_id().get_sha256();
                auto id = utils::bytes_t(sha256.begin(), sha256.end());
                auto msg = proto::relay::session_invitation_t{std::move(id), session_key, {}, 12345, true};
                send_relay(msg);
            }
        }

        outcome::result<void> operator()(const model::diff::contact::unknown_connected_t &diff,
                                         void *) noexcept override {
            CHECK(diff.device_id == peer_device->device_id());
            unknownon_connected = true;
            io_ctx.stop();
            return outcome::success();
        }

        bool sent = false;
        bool unknownon_connected = false;
    };
    F().run();
}

int _init() {
    REGISTER_TEST_CASE(test_master_connect, "test_master_connect", "[relay]");
    REGISTER_TEST_CASE(test_passive, "test_passive", "[relay]");
    REGISTER_TEST_CASE(test_passive_ping_pong, "test_passive_ping_pong", "[relay]");
    REGISTER_TEST_CASE(test_passive_unknown, "test_passive_unknown", "[relay]");

    return 1;
}

static int v = _init();

repo_name: basiliscos/syncspirit

license: mit


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#include "test-utils.h"
#include "access.h"

#include "utils/tls.h"
#include "utils/format.hpp"
#include "proto/bep_support.h"
#include "model/cluster.h"
#include "model/messages.h"
#include "model/diff/cluster_visitor.h"
#include "net/names.h"
#include "net/messages.h"
#include "net/peer_actor.h"
#include "transport/stream.h"
#include "diff-builder.h"
#include "constants.h"

#include <rotor/asio.hpp>
#include <boost/algorithm/string/replace.hpp>

using namespace syncspirit;
using namespace syncspirit::test;
using namespace syncspirit::model;
using namespace syncspirit::net;
using namespace std::chrono_literals;

namespace asio = boost::asio;
namespace sys = boost::system;
namespace r = rotor;
namespace ra = r::asio;

using configure_callback_t = std::function<void(r::plugin::plugin_base_t &)>;

auto timeout = r::pt::time_duration{r::pt::millisec{1500}};
auto host = "127.0.0.1";

struct supervisor_t : ra::supervisor_asio_t {
    using parent_t = ra::supervisor_asio_t;
    using parent_t::parent_t;

    void configure(r::plugin::plugin_base_t &plugin) noexcept override {
        parent_t::configure(plugin);
        plugin.with_casted<r::plugin::registry_plugin_t>([&](auto &p) {
            p.register_name(names::coordinator, get_address());
            p.register_name(names::peer_supervisor, get_address());
        });
        if (configure_callback) {
            configure_callback(plugin);
        }
    }

    void on_child_shutdown(actor_base_t *actor) noexcept override {
        if (actor) {
            spdlog::info("child shutdown: {}, reason: {}", actor->get_identity(),
                         actor->get_shutdown_reason()->message());
        }
        parent_t::on_child_shutdown(actor);
    }

    void shutdown_finish() noexcept override {
        parent_t::shutdown_finish();
        if (acceptor) {
            acceptor->cancel();
        }
    }

    auto get_state() noexcept { return state; }

    asio::ip::tcp::acceptor *acceptor = nullptr;
    configure_callback_t configure_callback;
};

using supervisor_ptr_t = r::intrusive_ptr_t<supervisor_t>;
using actor_ptr_t = r::intrusive_ptr_t<peer_actor_t>;

struct fixture_t : private model::diff::cluster_visitor_t {
    using acceptor_t = asio::ip::tcp::acceptor;
    using timer_t = asio::deadline_timer;
    using timer_ptr_t = std::unique_ptr<timer_t>;

    fixture_t() noexcept : ctx(io_ctx), acceptor(io_ctx), peer_sock(io_ctx), pre_main_counter{2} {
        test::init_logging();
        log = utils::get_logger("fixture");
    }

    virtual void run(bool add_peer = true) noexcept {
        known_peer = add_peer;
        auto strand = std::make_shared<asio::io_context::strand>(io_ctx);
        sup = ctx.create_supervisor<supervisor_t>().strand(strand).timeout(timeout).create_registry().finish();
        sup->configure_callback = [&](r::plugin::plugin_base_t &plugin) {
            plugin.template with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
                using cluster_diff_ptr_t = r::intrusive_ptr_t<model::message::model_update_t>;
                using cluster_diff_t = typename cluster_diff_ptr_t::element_type;

                p.subscribe_actor(r::lambda<cluster_diff_t>([&](cluster_diff_t &msg) {
                    LOG_INFO(log, "received cluster diff message");
                    auto &diff = msg.payload.diff;
                    auto r = diff->apply(*cluster, get_apply_controller());
                    if (!r) {
                        LOG_ERROR(log, "error updating model: {}", r.assume_error().message());
                        sup->do_shutdown();
                    }
                }));
            });
        };
        sup->start();
        sup->do_process();

        my_keys = utils::generate_pair("my").value();
        auto md = model::device_id_t::from_cert(my_keys.cert_data).value();
        my_device = device_t::create(md, "my-device").value();

        peer_keys = utils::generate_pair("peer").value();
        auto pd = model::device_id_t::from_cert(peer_keys.cert_data).value();
        peer_device = device_t::create(pd, "peer-device").value();

        cluster = new cluster_t(my_device, 1);
        cluster->get_devices().put(my_device);
        if (add_peer) {
            cluster->get_devices().put(peer_device);
        }

        auto ep = asio::ip::tcp::endpoint(asio::ip::make_address(host), 0);
        acceptor.open(ep.protocol());
        acceptor.bind(ep);
        acceptor.listen();
        auto local_ep = acceptor.local_endpoint();

        acceptor.async_accept(peer_sock, [this](auto ec) { this->accept(ec); });
        sup->acceptor = &acceptor;

        auto uri_str = fmt::format("tcp://{}:{}/", local_ep.address(), local_ep.port());
        LOG_TRACE(log, "Connecting to {}", uri_str);

        auto uri = utils::parse(uri_str);
        auto cfg = transport::transport_config_t{{}, uri, *sup, {}, true};
        client_trans = transport::initiate_stream(cfg);

        auto ip = asio::ip::make_address(host);
        auto peer_ep = tcp::endpoint(ip, local_ep.port());
        auto addresses = std::vector<tcp::endpoint>{peer_ep};
        auto addresses_ptr = std::make_shared<decltype(addresses)>(addresses);

        transport::error_fn_t on_error = [&](auto &ec) { on_client_error(ec); };
        transport::connect_fn_t on_connect = [addresses_ptr, this](const tcp::endpoint &) {
            LOG_INFO(log, "active/connected");
            try_main();
        };

        client_trans->async_connect(addresses_ptr, on_connect, on_error);

        std::this_thread::sleep_for(1ms);
        io_ctx.run();
        CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
    }

    void try_main() {
        --pre_main_counter;
        LOG_DEBUG(log, "try_main, counter = {}", pre_main_counter);
        if (!pre_main_counter) {
            main();
        }
    }

    virtual void main() noexcept {}

    virtual actor_ptr_t create_actor(uint32_t rx_timeout = 2000) noexcept {
        if (known_peer) {
            auto builder = diff_builder_t(*cluster);
            builder.update_state(*peer_device, {}, model::device_state_t::connecting).apply(*sup);
        }

        auto bep_config = config::bep_config_t();
        bep_config.rx_buff_size = 1024;
        bep_config.rx_timeout = rx_timeout;
        LOG_INFO(log, "crearing actor, timeout = {}", rx_timeout);

        return sup->create_actor<actor_ptr_t::element_type>()
            .timeout(timeout)
            .cluster(cluster)
            .coordinator(sup->get_address())
            .bep_config(bep_config)
            .transport(peer_trans)
            .peer_device_id(peer_device->device_id())
            .device_name("peer-device")
            .peer_proto("tcp")
            .autoshutdown_supervisor(true)
            .finish();
    }

    virtual void on_client_error(const sys::error_code &ec) noexcept { LOG_WARN(log, "client err: {}", ec.message()); }

    virtual void accept(const sys::error_code &ec) noexcept {
        LOG_INFO(log, "accept, ec: {}, remote = {}", ec.message(), peer_sock.remote_endpoint());
        auto uri = utils::parse("tcp://127.0.0.1:0/");
        auto cfg = transport::transport_config_t{{}, uri, *sup, std::move(peer_sock), false};
        peer_trans = transport::initiate_stream(cfg);
        try_main();
    }

    virtual void on_client_write(std::size_t bytes) noexcept { LOG_INFO(log, "client sent {} bytes", bytes); }

    virtual void on_client_read(std::size_t bytes) noexcept {
        LOG_INFO(log, "client received {} bytes", bytes);
        auto result = proto::parse_bep(utils::bytes_view_t(rx_buff, bytes)).value();
        auto hello = std::get_if<proto::Hello>(&result.message);
        // On some platforms 'on_client_read' triggers before 'on_server_send', but as we
        // check server code here, there might be problems. So, give server-side
        // some time via timer.
        if (hello) {
            assert(!hello_timer);
            hello_timer.reset(new timer_t(ctx.get_io_context()));
            hello_timer->expires_from_now(r::pt::milliseconds{1});
            hello_timer->async_wait([hello = std::move(*hello), this](const sys::error_code &ec) mutable {
                LOG_INFO(log, "on hello timer, ec: {}", ec.value());
                on_hello(hello);
                hello_timer.reset();
            });
        }
    }

    virtual void on_hello(proto::Hello &msg) noexcept {
        auto device_name = proto::get_device_name(msg);
        auto client_name = proto::get_client_name(msg);
        auto client_version = proto::get_client_version(msg);
        LOG_INFO(log, "client received hello message from {}, {}/{}", device_name, client_name, client_version);
    }

    virtual void send_hello() noexcept {
        tx_buff = proto::make_hello_message("self-name");
        transport::io_fn_t on_write = [&](size_t bytes) { on_client_write(bytes); };
        transport::io_fn_t on_read = [&](size_t bytes) { on_client_read(bytes); };
        transport::error_fn_t on_error = [&](auto &ec) { on_client_error(ec); };
        auto tx_buff_ = asio::buffer(tx_buff.data(), tx_buff.size());
        auto rx_buff_ = asio::buffer(rx_buff, sizeof(rx_buff));
        client_trans->async_recv(rx_buff_, on_read, on_error);
        client_trans->async_send(tx_buff_, on_write, on_error);
    }

    cluster_ptr_t cluster;
    supervisor_ptr_t sup;
    asio::io_context io_ctx;
    ra::system_context_asio_t ctx;
    acceptor_t acceptor;
    asio::ip::tcp::socket peer_sock;
    utils::logger_t log;
    utils::key_pair_t peer_keys;
    utils::key_pair_t my_keys;
    model::device_ptr_t peer_device;
    model::device_ptr_t my_device;
    transport::stream_sp_t peer_trans;
    transport::stream_sp_t client_trans;
    utils::bytes_t tx_buff;
    timer_ptr_t hello_timer;
    unsigned char rx_buff[2000];
    bool known_peer = false;
    int pre_main_counter;
};

void test_shutdown_on_hello_timeout() {
    struct F : fixture_t {
        void main() noexcept override { auto act = create_actor(1); }

        void run(bool add_peer = true) noexcept override {
            fixture_t::run(add_peer);
            CHECK(sup->get_state() == r::state_t::SHUT_DOWN);
        }
    };
    F().run();
}

void test_online_on_hello() {
    struct F : fixture_t {
        void main() noexcept override {
            create_actor();
            send_hello();
        }

        void on_hello(proto::Hello &) noexcept override {
            auto peer = cluster->get_devices().by_sha256(peer_device->device_id().get_sha256());
            CHECK(peer->get_state() == device_state_t::online);
        }
    };
    F().run();
}

void test_hello_from_unknown() {
    struct F : fixture_t {
        void main() noexcept override {
            create_actor();
            send_hello();
        }

        void on_hello(proto::Hello &) noexcept override {
            CHECK(cluster->get_devices().size() == 1);
            auto &unknown_devices = cluster->get_pending_devices();
            CHECK(unknown_devices.size() == 1);
            auto peer = unknown_devices.by_sha256(peer_device->device_id().get_sha256());
            REQUIRE(peer);
            CHECK(peer->get_name() == "self-name");
            CHECK(peer->get_client_name() == constants::client_name);
            CHECK(peer->get_client_version() == constants::client_version);
            CHECK(peer->get_address() == "tcp://0.0.0.0:0");

            auto delta = pt::microsec_clock::local_time() - peer->get_last_seen();
            CHECK(delta.seconds() <= 2);
            CHECK(my_device->get_rx_bytes() > 0);
            CHECK(my_device->get_tx_bytes() > 0);
        }
    };
    F().run(false);
}

void test_hello_from_known_unknown() {
    struct F : fixture_t {
        void main() noexcept override {
            diff_builder_t(*cluster).add_unknown_device(peer_device->device_id(), {}).apply(*sup);
            REQUIRE(cluster->get_pending_devices().size() == 1);
            create_actor();
            send_hello();
        }

        void on_hello(proto::Hello &) noexcept override {
            CHECK(cluster->get_devices().size() == 1);
            auto &unknown_devices = cluster->get_pending_devices();
            CHECK(unknown_devices.size() == 1);
            auto peer = unknown_devices.by_sha256(peer_device->device_id().get_sha256());
            REQUIRE(peer);
            CHECK(peer->get_name() == "self-name");
            CHECK(peer->get_client_name() == constants::client_name);
            CHECK(peer->get_client_version() == constants::client_version);
            CHECK(peer->get_address() == "tcp://0.0.0.0:0");

            auto delta = pt::microsec_clock::local_time() - peer->get_last_seen();
            CHECK(delta.seconds() <= 2);
            CHECK(my_device->get_rx_bytes() > 0);
            CHECK(my_device->get_tx_bytes() > 0);
        }
    };
    F().run(false);
}

void test_hello_from_ignored() {
    struct F : fixture_t {
        void main() noexcept override {
            diff_builder_t(*cluster).add_ignored_device(peer_device->device_id(), {}).apply(*sup);
            REQUIRE(cluster->get_ignored_devices().size() == 1);
            create_actor();
            send_hello();
        }

        void on_hello(proto::Hello &) noexcept override {
            CHECK(cluster->get_devices().size() == 1);
            auto &ignored_devices = cluster->get_ignored_devices();
            CHECK(ignored_devices.size() == 1);
            auto peer = ignored_devices.by_sha256(peer_device->device_id().get_sha256());
            REQUIRE(peer);
            CHECK(peer->get_name() == "self-name");
            CHECK(peer->get_client_name() == constants::client_name);
            CHECK(peer->get_client_version() == constants::client_version);
            CHECK(peer->get_address() == "tcp://0.0.0.0:0");

            auto delta = pt::microsec_clock::local_time() - peer->get_last_seen();
            CHECK(delta.seconds() <= 2);
            REQUIRE(cluster->get_pending_devices().size() == 0);
        }
    };
    F().run(false);
}

int _init() {
    REGISTER_TEST_CASE(test_shutdown_on_hello_timeout, "test_shutdown_on_hello_timeout", "[peer]");
    REGISTER_TEST_CASE(test_online_on_hello, "test_online_on_hello", "[peer]");
    REGISTER_TEST_CASE(test_hello_from_unknown, "test_hello_from_unknown", "[peer]");
    REGISTER_TEST_CASE(test_hello_from_known_unknown, "test_hello_from_known_unknown", "[peer]");
    REGISTER_TEST_CASE(test_hello_from_ignored, "test_hello_from_ignored", "[peer]");
    return 1;
}

static int v = _init();


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#include <catch2/catch_all.hpp>
#include <rotor.hpp>
#include <rotor/asio.hpp>
#include "test-utils.h"
#include "utils/error_code.h"
#include "net/resolver_actor.h"
#include "utils/format.hpp"

using namespace syncspirit;
using namespace syncspirit::test;
using namespace syncspirit::net;
using namespace std::chrono_literals;

namespace asio = boost::asio;
namespace sys = boost::system;
namespace r = rotor;
namespace ra = r::asio;

using configure_callback_t = std::function<void(r::plugin::plugin_base_t &)>;
using response_ptr_t = r::intrusive_ptr_t<message::resolve_response_t>;
using Catch::Matchers::StartsWith;

auto timeout = r::pt::time_duration{r::pt::millisec{1000}};

struct my_supervisor_t : ra::supervisor_asio_t {
    using parent_t = ra::supervisor_asio_t;
    using parent_t::parent_t;
    using responses_t = std::vector<response_ptr_t>;

    void configure(r::plugin::plugin_base_t &plugin) noexcept override {
        ra::supervisor_asio_t::configure(plugin);
        plugin.with_casted<r::plugin::starter_plugin_t>(
            [&](auto &p) { p.subscribe_actor(&my_supervisor_t::on_resolve); });

        if (configure_callback) {
            configure_callback(plugin);
        }
    }

    void on_resolve(message::resolve_response_t &res) noexcept { responses.emplace_back(&res); }

    responses_t responses;
    configure_callback_t configure_callback;
};

using supervisor_ptr_t = r::intrusive_ptr_t<my_supervisor_t>;
using actor_ptr_t = r::intrusive_ptr_t<resolver_actor_t>;

struct fixture_t {

    fixture_t() : ctx(io_ctx), root_path{unique_path()}, path_quard{root_path}, remote_resolver{io_ctx} {
        test::init_logging();
        log = utils::get_logger("fixture");
        rx_buff.resize(1500);
    }

    virtual void main() noexcept = 0;

    void run() {
        auto strand = std::make_shared<asio::io_context::strand>(io_ctx);
        sup = ctx.create_supervisor<my_supervisor_t>().strand(strand).timeout(timeout).create_registry().finish();
        sup->start();
        sup->do_process();

        hosts_path = root_path / "hosts";
        hosts_path_str = hosts_path.string();

        auto ep = asio::ip::udp::endpoint(asio::ip::make_address("127.0.0.1"), 0);
        remote_resolver.open(ep.protocol());
        remote_resolver.bind(ep);

        auto local_ep = remote_resolver.local_endpoint();
        log->info("remote resolver: {}", local_ep);

        main();

        sup->do_shutdown();
        sup->do_process();
        io_ctx.run();
    }

    asio::io_context io_ctx{1};
    ra::system_context_asio_t ctx;
    bfs::path root_path;
    bfs::path hosts_path;
    std::string hosts_path_str;
    path_guard_t path_quard;
    utils::logger_t log;
    supervisor_ptr_t sup;
    actor_ptr_t resolver;
    udp_socket_t remote_resolver;
    udp::endpoint resolver_endpoint;
    fmt::memory_buffer rx_buff;
};

void test_local_resolver() {
    struct F : fixture_t {
        void main() noexcept override {

            write_file(hosts_path, "127.0.0.2 lclhst.localdomain lclhst\n");

            resolver = sup->create_actor<resolver_actor_t>()
                           .resolve_timeout(timeout / 2)
                           .hosts_path(hosts_path_str.c_str())
                           .server_addresses("127.0.0.1:1234")
                           .timeout(timeout)
                           .finish();

            sup->do_process();

            sup->request<payload::address_request_t>(resolver->get_address(), "lclhst", 123).send(timeout);
            sup->do_process();
            REQUIRE(sup->responses.size() == 1);

            auto results = sup->responses.at(0)->payload.res->results;
            REQUIRE(results.size() == 1);
            REQUIRE(results.at(0) == asio::ip::make_address("127.0.0.2"));

            // cache hit
            sup->request<payload::address_request_t>(resolver->get_address(), "lclhst", 123).send(timeout);
            sup->do_process();
            REQUIRE(sup->responses.size() == 2);

            results = sup->responses.at(1)->payload.res->results;
            REQUIRE(results.size() == 1);
            REQUIRE(results.at(0) == asio::ip::make_address("127.0.0.2"));
        }
    };
    F().run();
}

void test_success_resolver() {
    struct F : fixture_t {
        void main() noexcept override {
            auto local_port = remote_resolver.local_endpoint().port();
            write_file(hosts_path, "");

            auto buff = asio::buffer(rx_buff.data(), rx_buff.size());
            remote_resolver.async_receive_from(buff, resolver_endpoint, [&](sys::error_code ec, size_t bytes) -> void {
                log->info("received {} bytes from resolver, ec = {}", bytes, ec.value());
                const unsigned char reply[] = {0x0e, 0x51, 0x81, 0x80, 0x00, 0x01, 0x00, 0x01, 0x00, 0x00, 0x00,
                                               0x00, 0x06, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x03, 0x63, 0x6f,
                                               0x6d, 0x00, 0x00, 0x01, 0x00, 0x01, 0xc0, 0x0c, 0x00, 0x01, 0x00,
                                               0x01, 0x00, 0x00, 0x01, 0x02, 0x00, 0x04, 0x8e, 0xfa, 0xcb, 0x8e};
                auto reply_str = std::string_view(reinterpret_cast<const char *>(reply), sizeof(reply));
                auto buff = asio::buffer(reply_str.data(), reply_str.size());
                remote_resolver.async_send_to(buff, resolver_endpoint, [&](sys::error_code ec, size_t bytes) {
                    log->info("sent {} bytes to resolver, ec = {}", bytes, ec.value());
                });
            });

            resolver = sup->create_actor<resolver_actor_t>()
                           .resolve_timeout(timeout / 2)
                           .hosts_path(hosts_path_str.c_str())
                           .server_addresses(fmt::format("127.0.0.1:{}", local_port))
                           .timeout(timeout)
                           .finish();

            sup->do_process();

            sup->request<payload::address_request_t>(resolver->get_address(), "google.com", 80).send(timeout);
            io_ctx.run();
            REQUIRE(sup->responses.size() == 1);

            auto &results = sup->responses.at(0)->payload.res->results;
            REQUIRE(results.size() == 1);
            REQUIRE_THAT(results.at(0).to_string(), StartsWith("142.250."));
        }
    };
    F().run();
}

void test_success_ip() {
    struct F : fixture_t {
        void main() noexcept override {
            auto local_port = remote_resolver.local_endpoint().port();
            write_file(hosts_path, "");

            resolver = sup->create_actor<resolver_actor_t>()
                           .resolve_timeout(timeout / 2)
                           .hosts_path(hosts_path_str.c_str())
                           .server_addresses(fmt::format("127.0.0.1:{}", local_port))
                           .timeout(timeout)
                           .finish();

            sup->do_process();

            sup->request<payload::address_request_t>(resolver->get_address(), "127.0.0.1", 80).send(timeout);
            sup->do_process();

            REQUIRE(sup->responses.size() == 1);

            auto &results = sup->responses.at(0)->payload.res->results;
            REQUIRE(results.size() == 1);
            REQUIRE(results.at(0) == asio::ip::make_address("127.0.0.1"));
        }
    };
    F().run();
}

void test_success_ipv6() {
    struct F : fixture_t {
        void main() noexcept override {
            auto local_port = remote_resolver.local_endpoint().port();
            write_file(hosts_path, "");

            resolver = sup->create_actor<resolver_actor_t>()
                           .resolve_timeout(timeout / 2)
                           .hosts_path(hosts_path_str.c_str())
                           .server_addresses(fmt::format("127.0.0.1:{}", local_port))
                           .timeout(timeout)
                           .finish();

            sup->do_process();

            sup->request<payload::address_request_t>(resolver->get_address(),
                                                     "[fde8:6819:8685:4d00:be03:58ff:fe74:c854]", 80)
                .send(timeout);
            sup->do_process();

            REQUIRE(sup->responses.size() == 1);

            if (test::has_ipv6()) {
                auto &results = sup->responses.at(0)->payload.res->results;
                REQUIRE(results.size() == 1);
                REQUIRE(results.at(0) == asio::ip::make_address("fde8:6819:8685:4d00:be03:58ff:fe74:c854"));
            } else {
                auto &ee = sup->responses.at(0)->payload.ee;
                REQUIRE(ee);
                REQUIRE(ee->ec == utils::make_error_code(utils::error_code_t::cares_failure));
            }
        }
    };
    F().run();
}

void test_garbage() {
    struct F : fixture_t {
        void main() noexcept override {
            auto local_port = remote_resolver.local_endpoint().port();
            write_file(hosts_path, "");

            auto buff = asio::buffer(rx_buff.data(), rx_buff.size());
            remote_resolver.async_receive_from(buff, resolver_endpoint, [&](sys::error_code ec, size_t bytes) -> void {
                log->info("received {} bytes from resolver, ec = {}", bytes, ec.value());
                const unsigned char reply[] = {0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00};
                auto reply_str = std::string_view(reinterpret_cast<const char *>(reply), sizeof(reply));
                auto buff = asio::buffer(reply_str.data(), reply_str.size());
                remote_resolver.async_send_to(buff, resolver_endpoint, [&](sys::error_code ec, size_t bytes) {
                    log->info("sent {} bytes to resolver, ec = {}", bytes, ec.value());
                });
            });

            resolver = sup->create_actor<resolver_actor_t>()
                           .resolve_timeout(timeout / 2)
                           .hosts_path(hosts_path_str.c_str())
                           .server_addresses(fmt::format("127.0.0.1:{}", local_port))
                           .timeout(timeout)
                           .finish();

            sup->do_process();

            sup->request<payload::address_request_t>(resolver->get_address(), "google.com", 80).send(timeout);
            io_ctx.run();
            REQUIRE(sup->responses.size() == 1);

            auto &ee = sup->responses.at(0)->payload.ee;
            REQUIRE(ee);
            REQUIRE(ee->ec.value() == static_cast<int>(utils::error_code_t::cares_failure));
        }
    };
    F().run();
}

void test_multi_replies() {
    struct F : fixture_t {
        void main() noexcept override {
            auto local_port = remote_resolver.local_endpoint().port();
            write_file(hosts_path, "");

            auto buff = asio::buffer(rx_buff.data(), rx_buff.size());
            remote_resolver.async_receive_from(buff, resolver_endpoint, [&](sys::error_code ec, size_t bytes) -> void {
                log->info("received {} bytes from resolver, ec = {}", bytes, ec.value());
                const unsigned char reply[] = {
                    0x5e, 0x60, 0x81, 0x80, 0x00, 0x01, 0x00, 0x03, 0x00, 0x00, 0x00, 0x00, 0x06, 0x72, 0x65, 0x6c,
                    0x61, 0x79, 0x73, 0x09, 0x73, 0x79, 0x6e, 0x63, 0x74, 0x68, 0x69, 0x6e, 0x67, 0x03, 0x6e, 0x65,
                    0x74, 0x00, 0x00, 0x01, 0x00, 0x01, 0xc0, 0x0c, 0x00, 0x05, 0x00, 0x01, 0x00, 0x00, 0x05, 0x31,
                    0x00, 0x0d, 0x0a, 0x70, 0x61, 0x72, 0x2d, 0x6b, 0x38, 0x73, 0x2d, 0x76, 0x34, 0xc0, 0x13, 0xc0,
                    0x32, 0x00, 0x01, 0x00, 0x01, 0x00, 0x00, 0x00, 0x67, 0x00, 0x04, 0x33, 0x9f, 0x56, 0xd0, 0xc0,
                    0x32, 0x00, 0x01, 0x00, 0x01, 0x00, 0x00, 0x00, 0x67, 0x00, 0x04, 0x33, 0x9f, 0x4b, 0x11};
                auto reply_str = std::string_view(reinterpret_cast<const char *>(reply), sizeof(reply));
                auto buff = asio::buffer(reply_str.data(), reply_str.size());
                remote_resolver.async_send_to(buff, resolver_endpoint, [&](sys::error_code ec, size_t bytes) {
                    log->info("sent {} bytes to resolver, ec = {}", bytes, ec.value());
                });
            });

            resolver = sup->create_actor<resolver_actor_t>()
                           .resolve_timeout(timeout / 2)
                           .hosts_path(hosts_path_str.c_str())
                           .server_addresses(fmt::format("127.0.0.1:{}", local_port))
                           .timeout(timeout)
                           .finish();

            sup->do_process();

            sup->request<payload::address_request_t>(resolver->get_address(), "relays.syncthing.net", 80).send(timeout);
            io_ctx.run();
            REQUIRE(sup->responses.size() == 1);

            auto &results = sup->responses.at(0)->payload.res->results;
            REQUIRE(results.size() == 2);
            REQUIRE(results.at(0) == asio::ip::make_address("51.159.86.208"));
            REQUIRE(results.at(1) == asio::ip::make_address("51.159.75.17"));
        }
    };
    F().run();
}

void test_wrong() {
    struct F : fixture_t {
        void main() noexcept override {
            auto local_port = remote_resolver.local_endpoint().port();
            write_file(hosts_path, "");

            auto buff = asio::buffer(rx_buff.data(), rx_buff.size());
            remote_resolver.async_receive_from(buff, resolver_endpoint, [&](sys::error_code ec, size_t bytes) -> void {
                log->info("received {} bytes from resolver, ec = {}", bytes, ec.value());
                const unsigned char reply[] = {0x0e, 0x51, 0x81, 0x80, 0x00, 0x01, 0x00, 0x01, 0x00, 0x00, 0x00,
                                               0x00, 0x06, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x03, 0x63, 0x61,
                                               0x6d, 0x00, 0x00, 0x01, 0x00, 0x01, 0xc0, 0x0c, 0x00, 0x01, 0x00,
                                               0x01, 0x00, 0x00, 0x01, 0x02, 0x00, 0x04, 0x8e, 0xfa, 0xcb, 0x8e};
                auto reply_str = std::string_view(reinterpret_cast<const char *>(reply), sizeof(reply));
                auto buff = asio::buffer(reply_str.data(), reply_str.size());
                remote_resolver.async_send_to(buff, resolver_endpoint, [&](sys::error_code ec, size_t bytes) {
                    log->info("sent {} bytes to resolver", bytes, ec.value());
                });
            });

            resolver = sup->create_actor<resolver_actor_t>()
                           .resolve_timeout(timeout / 2)
                           .hosts_path(hosts_path_str.c_str())
                           .server_addresses(fmt::format("127.0.0.1:{}", local_port))
                           .timeout(timeout)
                           .finish();

            sup->do_process();

            sup->request<payload::address_request_t>(resolver->get_address(), "google.com", 80).send(timeout);
            io_ctx.run();
            REQUIRE(sup->responses.size() == 1);

            auto &ee = sup->responses.at(0)->payload.ee;
            REQUIRE(ee);
            REQUIRE(ee->ec.value() == static_cast<int>(utils::error_code_t::cares_failure));
        }
    };
    F().run();
}

void test_timeout() {
    struct F : fixture_t {
        void main() noexcept override {
            auto local_port = remote_resolver.local_endpoint().port();
            write_file(hosts_path, "");

            resolver = sup->create_actor<resolver_actor_t>()
                           .resolve_timeout(timeout / 2)
                           .hosts_path(hosts_path_str.c_str())
                           .server_addresses(fmt::format("127.0.0.1:{}", local_port))
                           .timeout(timeout)
                           .finish();

            sup->do_process();

            sup->request<payload::address_request_t>(resolver->get_address(), "google.com", 80).send(timeout);
            io_ctx.run();
            REQUIRE(sup->responses.size() == 1);

            auto &ee = sup->responses.at(0)->payload.ee;
            REQUIRE(ee);
            REQUIRE(ee->ec.value() == static_cast<int>(r::error_code_t::request_timeout));
        }
    };
    F().run();
}

void test_cancellation() {
    struct F : fixture_t {
        void main() noexcept override {
            auto local_port = remote_resolver.local_endpoint().port();
            write_file(hosts_path, "");

            resolver = sup->create_actor<resolver_actor_t>()
                           .resolve_timeout(timeout / 2)
                           .hosts_path(hosts_path_str.c_str())
                           .server_addresses(fmt::format("127.0.0.1:{}", local_port))
                           .timeout(timeout)
                           .finish();

            sup->do_process();

            auto resolver_address = resolver->get_address();
            auto request = sup->request<payload::address_request_t>(resolver_address, "google.com", 80).send(timeout);
            sup->send<message::resolve_cancel_t::payload_t>(resolver_address, request, sup->get_address());

            io_ctx.run();
            REQUIRE(sup->responses.size() == 1);

            auto &ee = sup->responses.at(0)->payload.ee;
            REQUIRE(ee);
            REQUIRE(ee->ec.value() == static_cast<int>(asio::error::operation_aborted));
        }
    };
    F().run();
}

int _init() {
    REGISTER_TEST_CASE(test_local_resolver, "test_local_resolver", "[resolver]");
    REGISTER_TEST_CASE(test_success_resolver, "test_success_resolver", "[resolver]");
    REGISTER_TEST_CASE(test_success_ip, "test_success_ip", "[resolver]");
    REGISTER_TEST_CASE(test_success_ipv6, "test_success_ipv6", "[resolver]");
    REGISTER_TEST_CASE(test_multi_replies, "test_multi_replies", "[resolver]");
    REGISTER_TEST_CASE(test_garbage, "test_garbage", "[resolver]");
    REGISTER_TEST_CASE(test_wrong, "test_wrong", "[resolver]");
    REGISTER_TEST_CASE(test_timeout, "test_timeout", "[resolver]");
    REGISTER_TEST_CASE(test_cancellation, "test_cancellation", "[resolver]");
    return 1;
}

static int v = _init();


code:
// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2019-2025 Ivan Baidakou

#include "test-utils.h"
#include "access.h"
#include "test_supervisor.h"
#include "diff-builder.h"
#include "model/cluster.h"
#include "fs/scan_scheduler.h"
#include "net/names.h"

using namespace syncspirit;
using namespace syncspirit::test;
using namespace syncspirit::model;

struct fixture_t {
    using target_ptr_t = r::intrusive_ptr_t<fs::scan_scheduler_t>;

    fixture_t() noexcept {
        test::init_logging();
        log = utils::get_logger("fixture");
    }

    void run() noexcept {
        auto my_id =
            device_id_t::from_string("KHQNO2S-5QSILRK-YX4JZZ4-7L77APM-QNVGZJT-EKU7IFI-PNEPBMY-4MXFMQD").value();
        my_device = device_t::create(my_id, "my-device").value();

        cluster = new cluster_t(my_device, 1);

        cluster->get_devices().put(my_device);

        r::system_context_t ctx;
        sup = ctx.create_supervisor<supervisor_t>().timeout(timeout).create_registry().finish();
        sup->cluster = cluster;
        sup->configure_callback = [&](r::plugin::plugin_base_t &plugin) {
            plugin.template with_casted<r::plugin::registry_plugin_t>(
                [&](auto &p) { p.register_name(net::names::fs_scanner, sup->get_address()); });
        };

        sup->start();
        sup->do_process();

        folder_id = "1234-5678";

        CHECK(static_cast<r::actor_base_t *>(sup.get())->access<to::state>() == r::state_t::OPERATIONAL);
        sup->do_process();

        target = sup->create_actor<fs::scan_scheduler_t>().timeout(timeout).cluster(cluster).finish();
        sup->do_process();

        REQUIRE(static_cast<r::actor_base_t *>(target.get())->access<to::state>() == r::state_t::OPERATIONAL);
        main();

        sup->do_process();
        sup->shutdown();
        sup->do_process();

        CHECK(static_cast<r::actor_base_t *>(sup.get())->access<to::state>() == r::state_t::SHUT_DOWN);
    }

    virtual void main() noexcept {}

    utils::logger_t log;
    r::pt::time_duration timeout = r::pt::millisec{10};
    r::intrusive_ptr_t<supervisor_t> sup;
    cluster_ptr_t cluster;
    device_ptr_t my_device;
    std::string folder_id;
    target_ptr_t target;
    model::folder_ptr_t folder;
};

void test_1_folder() {
    struct F : fixture_t {
        void main() noexcept override {
            auto db_folder = db::Folder();
            db::set_id(db_folder, folder_id);

            auto builder = diff_builder_t(*cluster);

            SECTION("zero rescan time => no scan") {
                builder.upsert_folder(db_folder).apply(*sup);
                auto folder = cluster->get_folders().by_id(folder_id);
                CHECK(!folder->is_scanning());
            }

            SECTION("non-zero rescan time") {
                db::set_rescan_interval(db_folder, 3600);
                builder.upsert_folder(db_folder).apply(*sup);

                auto folder = cluster->get_folders().by_id(folder_id);
                CHECK(folder->is_scanning());

                SECTION("scan start/finish") {
                    builder.scan_finish(folder_id).apply(*sup);
                    REQUIRE(!folder->is_scanning());

                    REQUIRE(sup->timers.size() == 1);
                    sup->do_invoke_timer((*sup->timers.begin())->request_id);
                    sup->do_process();
                    REQUIRE(folder->is_scanning());
                }

                SECTION("synchronization start/finish") {
                    builder.scan_finish(folder_id).synchronization_start(folder_id).scan_request(folder_id).apply(*sup);
                    REQUIRE(!folder->is_scanning());
                    REQUIRE(sup->timers.size() == 0);

                    builder.synchronization_finish(folder_id).apply(*sup);
                    REQUIRE(sup->timers.size() == 1);
                    sup->do_invoke_timer((*sup->timers.begin())->request_id);
                    sup->do_process();
                    REQUIRE(folder->is_scanning());
                }
            }

            SECTION("suspending") {
                db::set_rescan_interval(db_folder, 3600);
                builder.upsert_folder(db_folder).apply(*sup);

                auto folder = cluster->get_folders().by_id(folder_id);
                CHECK(folder->is_scanning());
                builder.suspend(*folder).scan_finish(folder_id).apply(*sup);
                REQUIRE(sup->timers.size() == 0);
            }
        }
    };
    F().run();
};

void test_2_folders() {
    struct F : fixture_t {
        void main() noexcept override {
            auto f1_id = "1111";
            auto f2_id = "2222";
            auto db_folder_1 = db::Folder();
            auto db_folder_2 = db::Folder();
            db::set_id(db_folder_1, f1_id);
            db::set_id(db_folder_2, f2_id);

            auto rescan_min = std::uint32_t(2000);
            db::set_rescan_interval(db_folder_1, 4000);
            db::set_rescan_interval(db_folder_2, rescan_min);

            auto builder = diff_builder_t(*cluster);
            builder.upsert_folder(db_folder_1)
                .upsert_folder(db_folder_2)
                .apply(*sup)
                .scan_finish(f2_id)
                .apply(*sup)
                .scan_finish(f1_id)
                .apply(*sup);

            REQUIRE(sup->timers.size() == 1);
            sup->do_invoke_timer((*sup->timers.begin())->request_id);
            sup->do_process();

            auto f1 = cluster->get_folders().by_id(f1_id);
            auto f2 = cluster->get_folders().by_id(f2_id);

            REQUIRE(!f1->is_scanning());
            REQUIRE(f2->is_scanning());

            auto at = r::pt::microsec_clock::local_time() + r::pt::seconds{rescan_min + 1};
            builder.scan_finish(f2_id, at).apply(*sup);

            REQUIRE(sup->timers.size() == 1);
            sup->do_invoke_timer((*sup->timers.begin())->request_id);
            sup->do_process();
            REQUIRE(f1->is_scanning());
            REQUIRE(!f2->is_scanning());

            f2->set_scan_finish({});
            builder.scan_finish(f1_id, at).scan_request(f2_id).apply(*sup);
            REQUIRE(!f1->is_scanning());
            REQUIRE(f2->is_scanning());
        }
    };
    F().run();
};

int _init() {
    REGISTER_TEST_CASE(test_1_folder, "test_1_folder", "[fs]");
    REGISTER_TEST_CASE(test_2_folders, "test_2_folders", "[fs]");
    return 1;
}

static int v = _init();