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

#include "db_actor.h"
#include "names.h"
#include "db/prefix.h"
#include "db/utils.h"
#include "db/error_code.h"
#include "model/diff/advance/advance.h"
#include "model/diff/contact/ignored_connected.h"
#include "model/diff/contact/peer_state.h"
#include "model/diff/contact/unknown_connected.h"
#include "model/diff/load/blocks.h"
#include "model/diff/load/devices.h"
#include "model/diff/load/file_infos.h"
#include "model/diff/load/folder_infos.h"
#include "model/diff/load/folders.h"
#include "model/diff/load/ignored_devices.h"
#include "model/diff/load/ignored_folders.h"
#include "model/diff/load/load_cluster.h"
#include "model/diff/load/pending_devices.h"
#include "model/diff/load/pending_folders.h"
#include "model/diff/modify/add_blocks.h"
#include "model/diff/modify/add_ignored_device.h"
#include "model/diff/modify/add_pending_device.h"
#include "model/diff/modify/add_pending_folders.h"
#include "model/diff/modify/remove_blocks.h"
#include "model/diff/modify/remove_files.h"
#include "model/diff/modify/remove_folder.h"
#include "model/diff/modify/remove_folder_infos.h"
#include "model/diff/modify/remove_ignored_device.h"
#include "model/diff/modify/remove_peer.h"
#include "model/diff/modify/remove_pending_device.h"
#include "model/diff/modify/remove_pending_folders.h"
#include "model/diff/modify/update_peer.h"
#include "model/diff/modify/upsert_folder.h"
#include "model/diff/modify/upsert_folder_info.h"
#include "model/diff/peer/update_folder.h"
#include "model/diff/peer/cluster_update.h"
#include "model/diff/cluster_visitor.h"
#include "model/misc/error_code.h"
#include <string_view>
#include <cstring>

#ifdef WIN32_LEAN_AND_MEAN
#include <malloc.h>
#else
#include <alloca.h>
#endif

namespace syncspirit::net {

namespace {
namespace resource {
r::plugin::resource_id_t db = 0;
}
} // namespace

#if 0
static void _my_log(MDBX_log_level_t loglevel, const char *function,int line, const char *fmt, va_list args) noexcept
{
    vprintf(fmt, args);
}
#endif

using folder_infos_set_t = std::unordered_set<const model::folder_info_t *>;

db_actor_t::db_actor_t(config_t &config)
    : r::actor_base_t{config}, env{nullptr}, db_dir{config.db_dir}, db_config{config.db_config},
      cluster{config.cluster} {
    // mdbx_module_handler({}, {}, {});
    // mdbx_setup_debug(MDBX_LOG_TRACE, MDBX_DBG_ASSERT, &_my_log);
    auto r = mdbx_env_create(&env);
    if (r != MDBX_SUCCESS) {
        auto log = utils::get_logger("net.db");
        LOG_CRITICAL(log, "mdbx environment creation error ({}): {}", r, mdbx_strerror(r));
        throw std::runtime_error(std::string(mdbx_strerror(r)));
    }
}

db_actor_t::~db_actor_t() {
    if (env) {
        mdbx_env_close(env);
    }
}

void db_actor_t::configure(r::plugin::plugin_base_t &plugin) noexcept {
    r::actor_base_t::configure(plugin);
    plugin.with_casted<r::plugin::address_maker_plugin_t>([&](auto &p) {
        p.set_identity("net.db", false);
        log = utils::get_logger(identity);
    });
    plugin.with_casted<r::plugin::registry_plugin_t>([&](auto &p) {
        p.discover_name(names::coordinator, coordinator, false).link(false).callback([&](auto phase, auto &ee) {
            if (!ee && phase == r::plugin::registry_plugin_t::phase_t::linking) {
                auto p = get_plugin(r::plugin::starter_plugin_t::class_identity);
                auto plugin = static_cast<r::plugin::starter_plugin_t *>(p);
                plugin->subscribe_actor(&db_actor_t::on_model_load_release);
                plugin->subscribe_actor(&db_actor_t::on_model_update, coordinator);
                plugin->subscribe_actor(&db_actor_t::on_db_info, coordinator);
            }
        });
    });
    plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
        open();
        p.subscribe_actor(&db_actor_t::on_cluster_load);
    });
}

void db_actor_t::open() noexcept {
    resources->acquire(resource::db);
    auto &my_device = cluster->get_device();
    auto upper_limit = db_config.upper_limit;
    /* enable automatic size management */
    LOG_INFO(log, "open, db upper limit = {}", upper_limit);
    auto r = mdbx_env_set_geometry(env, -1, -1, upper_limit, -1, -1, -1);
    if (r != MDBX_SUCCESS) {
        LOG_ERROR(log, "open, mdbx set geometry error ({}): {}", r, mdbx_strerror(r));
        resources->release(resource::db);
        return do_shutdown(make_error(db::make_error_code(r)));
    }

    auto flags = MDBX_WRITEMAP | MDBX_LIFORECLAIM | MDBX_EXCLUSIVE | MDBX_NOSTICKYTHREADS | MDBX_SAFE_NOSYNC;
#if defined(WIN32) || defined(_WIN32) || defined(__WIN32)
    auto db_dir_w = db_dir.wstring();
    r = mdbx_env_openW(env, db_dir_w.c_str(), flags, 0664);
#else
    r = mdbx_env_open(env, db_dir.c_str(), flags, 0664);
#endif
    if (r != MDBX_SUCCESS) {
        LOG_ERROR(log, "open, mdbx open environment error ({}): {}, path: {}", r, mdbx_strerror(r), db_dir.string());
        resources->release(resource::db);
        return do_shutdown(make_error(db::make_error_code(r)));
    }
    auto txn = db::make_transaction(db::transaction_type_t::RO, env);
    if (!txn) {
        LOG_ERROR(log, "open, cannot create transaction {}", txn.error().message());
        resources->release(resource::db);
        return do_shutdown(make_error(db::make_error_code(r)));
    }

    auto db_ver = db::get_version(txn.value());
    if (!db_ver) {
        LOG_ERROR(log, "open, cannot get db version :: {}", db_ver.error().message());
        resources->release(resource::db);
        return do_shutdown(make_error(db::make_error_code(r)));
    }
    auto version = db_ver.value();
    LOG_DEBUG(log, "got db version: {}, expected : {} ", version, db::version);

    if (!txn) {
        LOG_ERROR(log, "open, cannot create transaction {}", txn.error().message());
        resources->release(resource::db);
        return do_shutdown(make_error(db::make_error_code(r)));
    }
    if (db_ver.value() != db::version) {
        txn = db::make_transaction(db::transaction_type_t::RW, txn.value());
        auto r = db::migrate(version, my_device, txn.value());
        if (!r) {
            LOG_ERROR(log, "open, cannot migrate db {}", r.error().message());
            resources->release(resource::db);
            return do_shutdown(make_error(r.error()));
        }
        LOG_INFO(log, "open, successfully migrated db: {} -> {} ", version, db::version);
    }
    resources->release(resource::db);
}

auto db_actor_t::get_txn() noexcept -> outcome::result<db::transaction_t *> {
    if (!txn_holder) {
        auto txn = db::make_transaction(db::transaction_type_t::RW, env);
        if (!txn) {
            return txn.assume_error();
        }
        txn_holder.reset(new db::transaction_t(std::move(txn.assume_value())));
        uncommitted = 0;
    }
    return txn_holder.get();
}

auto db_actor_t::commit_on_demand() noexcept -> outcome::result<void> {
    if (txn_holder && (++uncommitted >= db_config.uncommitted_threshold)) {
        LOG_DEBUG(log, "committing tx");
        auto r = txn_holder->commit();
        txn_holder.reset();
        return r;
    }
    return outcome::success();
}

auto db_actor_t::force_commit() noexcept -> outcome::result<void> {
    uncommitted = db_config.uncommitted_threshold;
    return outcome::success();
}

void db_actor_t::on_start() noexcept {
    r::actor_base_t::on_start();
    LOG_TRACE(log, "on_start");
}

void db_actor_t::shutdown_finish() noexcept {
    if (txn_holder) {
        auto r = commit_on_demand();
        if (!r) {
            auto &err = r.assume_error();
            LOG_ERROR(log, "cannot commit tx: {}", err.message());
        }
        txn_holder.reset();
    }
    auto r = mdbx_env_close(env);
    if (r != MDBX_SUCCESS) {
        LOG_ERROR(log, "open, mdbx close error ({}): {}", r, mdbx_strerror(r));
    }
    env = nullptr;
    r::actor_base_t::shutdown_finish();
}

void db_actor_t::on_db_info(message::db_info_request_t &request) noexcept {
    LOG_TRACE(log, "on_db_info");
    MDBX_stat stat;
    std::memset(&stat, 0, sizeof(stat));
    auto r = mdbx_env_stat_ex(env, nullptr, &stat, sizeof(stat));
    if (r != MDBX_SUCCESS) {
        LOG_ERROR(log, "mdbx_env_stat_ex, mdbx error ({}): {}", r, mdbx_strerror(r));
        auto ec = db::make_error_code(r);
        return reply_with_error(request, make_error(ec));
    }

    payload::db_info_response_t info{
        stat.ms_psize, stat.ms_depth, stat.ms_leaf_pages, stat.ms_overflow_pages, stat.ms_branch_pages, stat.ms_entries,
    };

    reply_to(request, info);
}

void db_actor_t::on_cluster_load(message::load_cluster_request_t &request) noexcept {
    LOG_TRACE(log, "on_cluster_load");
    using namespace model::diff;

    auto txn_opt = db::make_transaction(db::transaction_type_t::RO, env);
    if (!txn_opt) {
        return reply_with_error(request, make_error(txn_opt.error()));
    }
    auto &txn = txn_opt.value();

    auto devices_opt = db::load(db::prefix::device, txn);
    if (!devices_opt) {
        return reply_with_error(request, make_error(devices_opt.error()));
    }

    auto blocks_opt = db::load(db::prefix::block_info, txn);
    if (!blocks_opt) {
        return reply_with_error(request, make_error(blocks_opt.error()));
    }
    auto blocks = std::move(blocks_opt.value());

    auto ignored_devices_opt = db::load(db::prefix::ignored_device, txn);
    if (!ignored_devices_opt) {
        return reply_with_error(request, make_error(ignored_devices_opt.error()));
    }

    auto ignored_folders_opt = db::load(db::prefix::ignored_folder, txn);
    if (!ignored_folders_opt) {
        return reply_with_error(request, make_error(ignored_folders_opt.error()));
    }

    auto folders_opt = db::load(db::prefix::folder, txn);
    if (!folders_opt) {
        return reply_with_error(request, make_error(folders_opt.error()));
    }

    auto folder_infos_opt = db::load(db::prefix::folder_info, txn);
    if (!folder_infos_opt) {
        return reply_with_error(request, make_error(folder_infos_opt.error()));
    }

    auto file_infos_opt = db::load(db::prefix::file_info, txn);
    if (!file_infos_opt) {
        return reply_with_error(request, make_error(file_infos_opt.error()));
    }
    auto files = std::move(file_infos_opt.value());

    auto pending_devices_opt = db::load(db::prefix::pending_device, txn);
    if (!pending_devices_opt) {
        return reply_with_error(request, make_error(pending_devices_opt.error()));
    }

    auto pending_folders_opt = db::load(db::prefix::pending_folder, txn);
    if (!pending_folders_opt) {
        return reply_with_error(request, make_error(pending_folders_opt.error()));
    }

    auto diff = model::diff::cluster_diff_ptr_t{};
    diff.reset(new load::load_cluster_t(std::move(txn), blocks.size(), files.size()));

    auto current = diff->assign_child(new load::devices_t(std::move(devices_opt.value())));
    if (blocks.size()) {
        auto max_blocks = db_config.max_blocks_per_diff;
        auto ptr = blocks.data();
        auto end = ptr + blocks.size();
        while (ptr < end) {
            auto chunk_end = std::min(ptr + max_blocks, end);
            auto slice = decltype(blocks)(ptr, chunk_end);
            ptr = chunk_end;
            current = current->assign_sibling(new load::blocks_t(std::move(slice)));
        }
    }

    current = current->assign_sibling(new load::ignored_devices_t(std::move(ignored_devices_opt.value())))
                  ->assign_sibling(new load::ignored_folders_t(std::move(ignored_folders_opt.value())))
                  ->assign_sibling(new load::folders_t(std::move(folders_opt.value())))
                  ->assign_sibling(new load::folder_infos_t(std::move(folder_infos_opt.value())));

    if (files.size()) {
        auto max_files = db_config.max_files_per_diff;
        auto ptr = files.data();
        auto end = ptr + files.size();
        while (ptr < end) {
            auto chunk_end = std::min(ptr + max_files, end);
            auto count = chunk_end - ptr;
            auto items = load::file_infos_t::container_t();
            items.reserve(count);
            while (ptr != chunk_end) {
                using item_t = decltype(items)::value_type;
                auto key = ptr->key;
                auto data = ptr->value;
                auto db_fi = db::FileInfo();
                if (auto left = db::decode(ptr->value, db_fi); left) {
                    auto ec = make_error_code(model::error_code_t::file_info_deserialization_failure);
                    return reply_with_error(request, make_error(ec));
                }
                items.emplace_back(item_t{key, std::move(db_fi)});
                ++ptr;
            }
            current = current->assign_sibling(new load::file_infos_t(std::move(items)));
        }
    }

    current->assign_sibling(new load::pending_devices_t(std::move(pending_devices_opt.value())))
        ->assign_sibling(new load::pending_folders_t(std::move(pending_folders_opt.value())));

    reply_to(request, diff);
}

void db_actor_t::on_model_load_release(model::message::model_update_t &message) noexcept {
    struct custom_visitor_t : model::diff::cluster_visitor_t {
        outcome::result<void> operator()(const model::diff::load::load_cluster_t &diff,
                                         void *custom) noexcept override {
            auto self = static_cast<db_actor_t *>(custom);
            LOG_DEBUG(self->log, "closing load transaction");
            auto &txn = const_cast<db::transaction_t &>(diff.txn);
            return txn.commit();
        }
    };

    auto visitor = custom_visitor_t();
    LOG_TRACE(log, "on_model_load_release", identity);
    auto r = message.payload.diff->visit(visitor, this);
    if (!r) {
        auto ee = make_error(r.assume_error());
        LOG_ERROR(log, "on_model_update error: {}", r.assume_error().message());
        do_shutdown(ee);
    }
}

void db_actor_t::on_model_update(model::message::model_update_t &message) noexcept {
    LOG_TRACE(log, "on_model_update", identity);
    auto &diff = *message.payload.diff;
    auto set = folder_infos_set_t{};
    auto r = diff.visit(*this, &set);
    while (r && !set.empty()) {
        r = save_folder_info(**set.begin(), &set);
    }

    if (r) {
        r = commit_on_demand();
    }
    if (!r) {
        auto ee = make_error(r.assume_error());
        LOG_ERROR(log, "on_model_update error: {}", r.assume_error().message());
        do_shutdown(ee);
    }
}

auto db_actor_t::save_folder_info(const model::folder_info_t &folder_info, void *custom) noexcept
    -> outcome::result<void> {
    auto txn_opt = get_txn();
    if (!txn_opt) {
        return txn_opt.assume_error();
    }
    auto &txn = *txn_opt.assume_value();

    auto fi_key = folder_info.get_key();
    auto fi_data = folder_info.serialize();
    auto r = db::save({fi_key, fi_data}, txn);

    if (!r) {
        return r;
    }

    auto folder_infos = reinterpret_cast<folder_infos_set_t *>(custom);
    auto it = folder_infos->find(&folder_info);
    if (it != folder_infos->end()) {
        folder_infos->erase(it);
    }
    return force_commit();
}

auto db_actor_t::remove(const model::diff::modify::generic_remove_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }
    auto &txn = *get_txn().assume_value();

    for (auto &key : diff.keys) {
        auto r = db::remove(key, txn);
        if (!r) {
            return r.assume_error();
        }
    }

    auto r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }

    return force_commit();
}

auto db_actor_t::operator()(const model::diff::peer::cluster_update_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }

    auto &txn = *get_txn().assume_value();

    auto &pending = cluster->get_pending_folders();
    if (pending.size()) {
        for (auto &it : pending) {
            auto &uf = it.item;
            auto key = uf->get_key();
            auto data = uf->serialize();

            auto r = db::save({key, data}, txn);
            if (!r) {
                return r.assume_error();
            }
        }
    }

    auto r = diff.visit_next(*this, custom);
    if (r.has_error()) {
        return r.assume_error();
    }

    return force_commit();
}

auto db_actor_t::operator()(const model::diff::modify::upsert_folder_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }
    auto &txn = *get_txn().assume_value();

    auto &db = diff.db;
    auto folder = cluster->get_folders().by_id(db::get_id(db));
    assert(folder);
    auto f_key = folder->get_key();
    auto f_data = folder->serialize();

    auto r = db::save({f_key, f_data}, txn);
    if (!r) {
        return r.assume_error();
    }

    r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }

    return force_commit();
}

auto db_actor_t::operator()(const model::diff::modify::add_blocks_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }
    auto &txn = *get_txn().assume_value();

    auto &blocks_map = cluster->get_blocks();
    for (const auto &it : diff.blocks) {
        auto block = blocks_map.by_hash(proto::get_hash(it));
        auto key = block->get_key();
        auto data = block->serialize();
        auto r = db::save({key, data}, txn);
        if (!r) {
            return r.assume_error();
        }
    }

    auto r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }

    return force_commit();
}

auto db_actor_t::operator()(const model::diff::modify::add_pending_folders_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }
    auto &txn = *get_txn().assume_value();

    auto &pending = cluster->get_pending_folders();
    for (auto &item : diff.container) {
        auto &folder = db::get_folder(item.db);
        auto id = db::get_id(folder);
        auto uf = pending.by_id(id);
        if (uf && uf->device_id().get_sha256() == item.peer_id) {
            if (uf->get_id() == id) {
                auto key = uf->get_key();
                auto data = uf->serialize();

                auto r = db::save({key, data}, txn);
                if (!r) {
                    return r.assume_error();
                }
                break;
            }
        }
    }

    auto r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }

    return force_commit();
}

auto db_actor_t::operator()(const model::diff::modify::add_ignored_device_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }
    auto &txn = *get_txn().assume_value();

    auto r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }

    auto device = cluster->get_ignored_devices().by_sha256(diff.device_id.get_sha256());
    auto key = device->get_key();
    auto data = device->serialize();

    r = db::save({key, data}, txn);
    if (!r) {
        return r.assume_error();
    }

    return force_commit();
}

auto db_actor_t::operator()(const model::diff::modify::add_pending_device_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }
    auto &txn = *get_txn().assume_value();
    auto device = cluster->get_pending_devices().by_sha256(diff.device_id.get_sha256());

    auto key = device->get_key();
    auto data = device->serialize();

    auto r = db::save({key, data}, txn);
    if (!r) {
        return r.assume_error();
    }

    r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }
    return force_commit();
}

auto db_actor_t::operator()(const model::diff::modify::remove_blocks_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    return remove(static_cast<const model::diff::modify::generic_remove_t &>(diff), custom);
}

auto db_actor_t::operator()(const model::diff::modify::remove_files_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    return remove(static_cast<const model::diff::modify::generic_remove_t &>(diff), custom);
}

auto db_actor_t::operator()(const model::diff::modify::remove_folder_infos_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    return remove(static_cast<const model::diff::modify::generic_remove_t &>(diff), custom);
}

auto db_actor_t::operator()(const model::diff::modify::remove_pending_folders_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    return remove(static_cast<const model::diff::modify::generic_remove_t &>(diff), custom);
}

auto db_actor_t::operator()(const model::diff::modify::remove_folder_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    auto &txn = *get_txn().assume_value();

    auto r = db::remove(diff.folder_key, txn);
    if (!r) {
        return r.assume_error();
    }

    r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }

    return force_commit();
}

auto db_actor_t::operator()(const model::diff::modify::remove_peer_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }
    auto &txn = *get_txn().assume_value();

    auto r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }
    r = db::remove(diff.peer_key, txn);
    if (!r) {
        return r.assume_error();
    }

    return force_commit();
}

auto db_actor_t::operator()(const model::diff::modify::remove_ignored_device_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }
    auto &txn = *get_txn().assume_value();

    auto r = db::remove(diff.device_key, txn);
    if (!r) {
        return r.assume_error();
    }

    r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }

    return force_commit();
}

auto db_actor_t::operator()(const model::diff::modify::remove_pending_device_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }
    auto &txn = *get_txn().assume_value();

    auto r = db::remove(diff.device_key, txn);
    if (!r) {
        return r.assume_error();
    }

    r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }

    return force_commit();
}

auto db_actor_t::operator()(const model::diff::modify::update_peer_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }

    auto &device_id = diff.peer_id;
    auto device = cluster->get_devices().by_sha256(device_id);
    assert(device);

    auto key = device->get_key();
    auto data = device->serialize();
    auto &txn = *get_txn().assume_value();

    auto r = db::save({key, data}, txn);
    if (!r) {
        return r.assume_error();
    }

    r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }

    return force_commit();
}

auto db_actor_t::operator()(const model::diff::modify::upsert_folder_info_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }

    auto device = cluster->get_devices().by_sha256(diff.device_id);
    auto folder = cluster->get_folders().by_id(diff.folder_id);
    auto folder_info = folder->get_folder_infos().by_device(*device);

    auto r = save_folder_info(*folder_info, custom);
    if (!r) {
        return r.assume_error();
    }

    r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }

    return r;
}

auto db_actor_t::operator()(const model::diff::advance::advance_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }

    auto folder = cluster->get_folders().by_id(diff.folder_id);
    if (folder && !folder->is_suspended()) {
        auto folder_info = folder->get_folder_infos().by_device(*cluster->get_device());
        if (folder_info) {
            auto name = proto::get_name(diff.proto_local);
            auto file = folder_info->get_file_infos().by_name(name);
            auto &txn = *get_txn().assume_value();

            {
                auto key = file->get_key();
                auto data = file->serialize();
                auto r = db::save({key, data}, txn);
                if (!r) {
                    return r.assume_error();
                }
            }

            auto folder_infos = reinterpret_cast<folder_infos_set_t *>(custom);
            folder_infos->emplace(folder_info.get());

            auto r = diff.visit_next(*this, custom);
            if (!r) {
                return r.assume_error();
            }
        }
    }

    return force_commit();
}

auto db_actor_t::operator()(const model::diff::peer::update_folder_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }

    auto txn_opt = get_txn();
    if (!txn_opt) {
        return txn_opt.assume_error();
    }
    auto &txn = *txn_opt.assume_value();

    auto folder = cluster->get_folders().by_id(diff.folder_id);
    auto folder_info = folder->get_folder_infos().by_device_id(diff.peer_id);

    auto fi_key = folder_info->get_key();
    auto fi_data = folder_info->serialize();
    auto r = db::save({fi_key, fi_data}, txn);
    if (!r) {
        return r.assume_error();
    }

    auto &files_map = folder_info->get_file_infos();
    for (const auto &f : diff.files) {
        auto name = proto::get_name(f);
        auto file = files_map.by_name(name);
        LOG_TRACE(log, "saving {}, seq = {}", file->get_full_name(), file->get_sequence());
        auto key = file->get_key();
        auto data = file->serialize();
        auto r = db::save({key, data}, txn);
        if (!r) {
            return r.assume_error();
        }
    }

    r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }

    return force_commit();
}

auto db_actor_t::operator()(const model::diff::contact::peer_state_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }

    if (!diff.has_been_online) {
        return outcome::success();
    }

    auto &device_id = diff.peer_id;
    auto device = cluster->get_devices().by_sha256(device_id);
    assert(device);

    auto key = device->get_key();
    auto data = device->serialize();
    auto &txn = *get_txn().assume_value();

    auto r = db::save({key, data}, txn);
    if (!r) {
        return r.assume_error();
    }

    r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }

    return force_commit();
}

auto db_actor_t::operator()(const model::diff::contact::ignored_connected_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }

    auto device = cluster->get_ignored_devices().by_sha256(diff.device_id.get_sha256());
    assert(device);

    auto key = device->get_key();
    auto data = device->serialize();
    auto &txn = *get_txn().assume_value();

    auto r = db::save({key, data}, txn);
    if (!r) {
        return r.assume_error();
    }

    r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }

    return force_commit();
}

auto db_actor_t::operator()(const model::diff::contact::unknown_connected_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (cluster->is_tainted()) {
        return outcome::success();
    }

    auto device = cluster->get_pending_devices().by_sha256(diff.device_id.get_sha256());
    assert(device);

    auto key = device->get_key();
    auto data = device->serialize();
    auto &txn = *get_txn().assume_value();

    auto r = db::save({key, data}, txn);
    if (!r) {
        return r.assume_error();
    }

    r = diff.visit_next(*this, custom);
    if (!r) {
        return r.assume_error();
    }

    return force_commit();
}

} // namespace syncspirit::net


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

#pragma once

#include "messages.h"
#include "config/db.h"
#include "model/messages.h"
#include "model/cluster.h"
#include "model/diff/cluster_visitor.h"
#include "mdbx.h"
#include "utils/log.h"
#include "db/transaction.h"

namespace syncspirit {
namespace net {

namespace outcome = boost::outcome_v2;
namespace bfs = std::filesystem;

struct db_actor_config_t : r::actor_config_t {
    bfs::path db_dir;
    config::db_config_t db_config;
    model::cluster_ptr_t cluster;
    size_t uncommitted_threshold = {100};
};

template <typename Actor> struct db_actor_config_builder_t : r::actor_config_builder_t<Actor> {
    using builder_t = typename Actor::template config_builder_t<Actor>;
    using parent_t = r::actor_config_builder_t<Actor>;
    using parent_t::parent_t;

    builder_t &&db_dir(const bfs::path &value) && noexcept {
        parent_t::config.db_dir = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&cluster(const model::cluster_ptr_t &value) && noexcept {
        parent_t::config.cluster = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&db_config(const config::db_config_t &value) && noexcept {
        parent_t::config.db_config = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }
};

struct SYNCSPIRIT_API db_actor_t : public r::actor_base_t, private model::diff::cluster_visitor_t {
    using config_t = db_actor_config_t;
    template <typename Actor> using config_builder_t = db_actor_config_builder_t<Actor>;

    static void delete_tx(MDBX_txn *) noexcept;

    db_actor_t(config_t &config);
    ~db_actor_t();
    void configure(r::plugin::plugin_base_t &plugin) noexcept override;
    void on_start() noexcept override;
    void shutdown_finish() noexcept override;

    template <typename T> auto &access() noexcept;

  private:
    using transaction_ptr_t = std::unique_ptr<db::transaction_t>;

    void open() noexcept;
    outcome::result<db::transaction_t *> get_txn() noexcept;
    outcome::result<void> commit_on_demand() noexcept;
    outcome::result<void> force_commit() noexcept;

    void on_cluster_load(message::load_cluster_request_t &message) noexcept;
    void on_model_load_release(model::message::model_update_t &) noexcept;
    void on_model_update(model::message::model_update_t &) noexcept;
    void on_db_info(message::db_info_request_t &) noexcept;
    void extracted(const model::folder_info_t &folder_info);
    outcome::result<void> save_folder_info(const model::folder_info_t &, void *) noexcept;
    outcome::result<void> remove(const model::diff::modify::generic_remove_t &, void *) noexcept;

    outcome::result<void> operator()(const model::diff::advance::advance_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::contact::peer_state_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::contact::ignored_connected_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::contact::unknown_connected_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::add_blocks_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::add_ignored_device_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::add_pending_device_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::add_pending_folders_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::remove_blocks_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::remove_files_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::remove_folder_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::remove_folder_infos_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::remove_ignored_device_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::remove_peer_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::remove_pending_device_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::remove_pending_folders_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::update_peer_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::upsert_folder_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::upsert_folder_info_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::peer::cluster_update_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::peer::update_folder_t &, void *) noexcept override;

    r::address_ptr_t coordinator;
    utils::logger_t log;
    MDBX_env *env;
    bfs::path db_dir;
    config::db_config_t db_config;
    model::cluster_ptr_t cluster;
    transaction_ptr_t txn_holder;
    std::int_fast32_t uncommitted;
};

} // namespace net
} // namespace syncspirit

repo_name: basiliscos/syncspirit

license: mit


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

#include "dialer_actor.h"
#include "model/diff/modify/remove_peer.h"
#include "model/diff/contact/dial_request.h"
#include "model/diff/contact/peer_state.h"
#include "model/diff/contact/update_contact.h"
#include "names.h"
#include "utils/format.hpp"
#include <cassert>

namespace syncspirit::net {

using state_t = model::device_state_t;

namespace {
namespace resource {
r::plugin::resource_id_t timer = 0;
} // namespace resource
} // namespace

dialer_actor_t::dialer_actor_t(config_t &config)
    : r::actor_base_t{config}, cluster{config.cluster},
      redial_timeout{r::pt::milliseconds{config.dialer_config.redial_timeout}},
      skip_discovers{config.dialer_config.skip_discovers}, announced{false} {}

void dialer_actor_t::configure(r::plugin::plugin_base_t &plugin) noexcept {
    r::actor_base_t::configure(plugin);
    plugin.with_casted<r::plugin::address_maker_plugin_t>([&](auto &p) {
        p.set_identity("net.dialer", false);
        log = utils::get_logger(identity);
    });
    plugin.with_casted<r::plugin::registry_plugin_t>([&](auto &p) {
        p.discover_name(net::names::coordinator, coordinator, false).link(false).callback([&](auto phase, auto &ee) {
            if (!ee && phase == r::plugin::registry_plugin_t::phase_t::linking) {
                auto p = get_plugin(r::plugin::starter_plugin_t::class_identity);
                auto plugin = static_cast<r::plugin::starter_plugin_t *>(p);
                plugin->subscribe_actor(&dialer_actor_t::on_announce, coordinator);
                plugin->subscribe_actor(&dialer_actor_t::on_model_update, coordinator);
            }
        });
    });
}

void dialer_actor_t::on_start() noexcept {
    LOG_TRACE(log, "on_start");
    r::actor_base_t::on_start();
    auto &devices = cluster->get_devices();
    for (auto it : devices) {
        auto &d = it.item;
        if (d != cluster->get_device()) {
            schedule_redial(d);
        }
    }
}

void dialer_actor_t::shutdown_finish() noexcept {
    LOG_TRACE(log, "shutdown_finish");
    r::actor_base_t::shutdown_finish();
}

void dialer_actor_t::shutdown_start() noexcept {
    LOG_TRACE(log, "shutdown_start");
    r::actor_base_t::shutdown_start();
    while (!redial_map.empty()) {
        auto it = redial_map.begin();
        auto &info = it->second;
        if (info.timer_id) {
            cancel_timer(*info.timer_id);
        } else {
            redial_map.erase(it);
        }
    }
}

void dialer_actor_t::on_announce(message::announce_notification_t &) noexcept {
    LOG_TRACE(log, "on_announce");
    announced = true;
    auto &devices = cluster->get_devices();
    for (auto it : devices) {
        auto &d = it.item;
        if (d != cluster->get_device()) {
            schedule_redial(d);
        }
    }
}

void dialer_actor_t::discover_or_dial(const model::device_ptr_t &peer_device) noexcept {
    auto dynamic = peer_device->is_dynamic();
    if ((dynamic && !announced) || (state != r::state_t::OPERATIONAL)) {
        return;
    }

    auto &info = redial_map[peer_device];
    info.last_attempt = clock_t::now();
    auto &device_id = peer_device->device_id();
    bool do_dial = false;
    bool do_discover = false;
    bool do_update = false;

    if (!dynamic) {
        if (peer_device->get_uris().size()) {
            do_dial = true;
        } else {
            do_update = true;
        }
    } else {
        if (!peer_device->get_uris().size()) {
            do_discover = true;
        } else if (info.skip_discovers) {
            --info.skip_discovers;
            do_dial = true;
        } else {
            do_discover = true;
        }
    }

    if (do_discover) {
        auto diff =
            model::diff::contact::peer_state_t::create(*cluster, device_id.get_sha256(), nullptr, state_t::discovering);
        if (diff) {
            send<model::payload::model_update_t>(coordinator, std::move(diff));
        }
    }
    if (do_dial) {
        auto diff = new model::diff::contact::dial_request_t(*peer_device);
        send<model::payload::model_update_t>(coordinator, std::move(diff), this);
        LOG_TRACE(log, "discover_or_dial, dialing to {} (dynamic = {}, discover attempts left = {})", device_id,
                  dynamic, info.skip_discovers);
    }
    if (do_update) {
        // will dial indirectly
        auto &uris = peer_device->get_static_uris();
        auto diff = new model::diff::contact::update_contact_t(*cluster, device_id, uris);
        send<model::payload::model_update_t>(coordinator, std::move(diff), this);
    }
}

void dialer_actor_t::schedule_redial(const model::device_ptr_t &peer_device) noexcept {
    using ms_t = std::chrono::milliseconds;
    auto &info = redial_map[peer_device];
    if (info.timer_id.has_value()) {
        return;
    }

    auto delta = ms_t(redial_timeout.total_milliseconds());
    auto deadline = info.last_attempt + delta;
    auto now = clock_t::now();
    if (deadline >= now) {
        auto diff_ms = std::chrono::duration_cast<ms_t>(deadline - now).count();
        auto diff = pt::millisec(diff_ms);
        auto redial_timer = start_timer(diff, *this, &dialer_actor_t::on_timer);
        info.timer_id = redial_timer;
        LOG_TRACE(log, "scheduling (re)dial to {} in {} ms, timer_id = {}", peer_device->device_id(), diff_ms,
                  redial_timer);
        resources->acquire(resource::timer);
    } else {
        LOG_TRACE(log, "will discover '{}' immediately", peer_device->device_id());
        discover_or_dial(peer_device);
    }
}

void dialer_actor_t::on_timer(r::request_id_t request_id, bool cancelled) noexcept {
    LOG_TRACE(log, "on_timer, cancelled = {}, timer_id = {}", cancelled, request_id);
    using value_t = typename redial_map_t::value_type;
    resources->release(resource::timer);
    auto predicate = [&](const value_t &val) -> bool { return val.second.timer_id == request_id; };
    auto it = std::find_if(redial_map.begin(), redial_map.end(), predicate);
    assert(it != redial_map.end());
    auto &peer = it->first;
    auto &info = it->second;
    info.timer_id.reset();
    if (!cancelled) {
        discover_or_dial(peer);
    }
}

void dialer_actor_t::on_model_update(model::message::model_update_t &msg) noexcept {
    LOG_TRACE(log, "on_model_update");
    auto &diff = *msg.payload.diff;
    auto r = diff.visit(*this, nullptr);
    if (!r) {
        auto ee = make_error(r.assume_error());
        do_shutdown(ee);
    }
}

auto dialer_actor_t::operator()(const model::diff::contact::peer_state_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    auto &devices = cluster->get_devices();
    auto peer = devices.by_sha256(diff.peer_id);

    if (peer) {
        if (diff.state == state_t::offline) {
            schedule_redial(peer);
        } else if (diff.state == state_t::online) {
            auto it = redial_map.find(peer);
            if (it != redial_map.end()) {
                auto &info = it->second;
                if (info.timer_id) {
                    cancel_timer(*info.timer_id);
                }
            }
        }
    }

    return diff.visit_next(*this, custom);
}

auto dialer_actor_t::operator()(const model::diff::contact::update_contact_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    auto &devices = cluster->get_devices();
    auto peer = devices.by_sha256(diff.device.get_sha256());
    if (!peer || peer == cluster->get_device()) {
        return diff.visit_next(*this, custom);
    }

    using state_t = model::device_state_t;
    auto state = peer->get_state();
    if (peer->get_uris().size() && (state == state_t::offline || state == state_t::discovering)) {
        LOG_TRACE(log, "dialing to {}", peer->device_id());
        auto diff = new model::diff::contact::dial_request_t(*peer);
        send<model::payload::model_update_t>(coordinator, std::move(diff), this);
        if (state == state_t::discovering) {
            auto it = redial_map.find(peer);
            if (it != redial_map.end()) {
                it->second.skip_discovers = skip_discovers;
            }
        }
    }

    return diff.visit_next(*this, custom);
}

auto dialer_actor_t::operator()(const model::diff::modify::remove_peer_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    for (auto it = redial_map.begin(); it != redial_map.end(); ++it) {
        if (it->first->get_key() == diff.peer_key) {
            auto &info = it->second;
            if (info.timer_id.has_value()) {
                cancel_timer(*info.timer_id);
            }
            redial_map.erase(it);
            break;
        }
    }
    return diff.visit_next(*this, custom);
}

} // namespace syncspirit::net


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

#pragma once

#include "config/dialer.h"
#include "config/bep.h"
#include "utils/log.h"
#include "model/cluster.h"
#include "model/messages.h"
#include "model/diff/cluster_visitor.h"
#include "messages.h"
#include <optional>
#include <unordered_map>
#include <chrono>

namespace syncspirit {
namespace net {

namespace outcome = boost::outcome_v2;

struct dialer_actor_config_t : r::actor_config_t {
    config::dialer_config_t dialer_config;
    model::cluster_ptr_t cluster;
};

template <typename Actor> struct dialer_actor_config_builder_t : r::actor_config_builder_t<Actor> {
    using builder_t = typename Actor::template config_builder_t<Actor>;
    using parent_t = r::actor_config_builder_t<Actor>;
    using parent_t::parent_t;

    builder_t &&cluster(const model::cluster_ptr_t &value) && noexcept {
        parent_t::config.cluster = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&dialer_config(const config::dialer_config_t &value) && noexcept {
        parent_t::config.dialer_config = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }
};

struct SYNCSPIRIT_API dialer_actor_t : public r::actor_base_t, private model::diff::cluster_visitor_t {
    using config_t = dialer_actor_config_t;
    template <typename Actor> using config_builder_t = dialer_actor_config_builder_t<Actor>;

    explicit dialer_actor_t(config_t &config);
    void configure(r::plugin::plugin_base_t &plugin) noexcept override;
    void shutdown_start() noexcept override;
    void shutdown_finish() noexcept override;
    void on_start() noexcept override;

  private:
    using clock_t = std::chrono::steady_clock;
    using timer_option_t = std::optional<rotor::request_id_t>;
    struct redial_info_t {
        clock_t::time_point last_attempt;
        timer_option_t timer_id;
        std::size_t skip_discovers;
    };
    using redial_map_t = std::unordered_map<model::device_ptr_t, redial_info_t>;

    void on_announce(message::announce_notification_t &message) noexcept;
    void on_model_update(model::message::model_update_t &) noexcept;

    void discover_or_dial(const model::device_ptr_t &device) noexcept;
    void remove(const model::device_ptr_t &device) noexcept;
    void schedule_redial(const model::device_ptr_t &device) noexcept;
    void on_timer(r::request_id_t request_id, bool cancelled) noexcept;

    outcome::result<void> operator()(const model::diff::contact::peer_state_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::contact::update_contact_t &, void *) noexcept override;
    outcome::result<void> operator()(const model::diff::modify::remove_peer_t &, void *) noexcept override;

    utils::logger_t log;
    model::cluster_ptr_t cluster;
    pt::time_duration redial_timeout;
    std::uint32_t skip_discovers;

    r::address_ptr_t coordinator;
    redial_map_t redial_map;
    bool announced;
};

} // namespace net
} // namespace syncspirit


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

#include "global_discovery_actor.h"
#include "names.h"
#include <nlohmann/json.hpp>
#include "proto/discovery_support.h"
#include "utils/beast_support.h"
#include "utils/error_code.h"
#include "http_actor.h"
#include "model/diff/contact/update_contact.h"
#include "model/diff/contact/peer_state.h"
#include "utils/format.hpp"

using namespace syncspirit::net;

namespace {
namespace resource {
r::plugin::resource_id_t timer = 0;
r::plugin::resource_id_t http = 1;
} // namespace resource
} // namespace

global_discovery_actor_t::global_discovery_actor_t(config_t &cfg)
    : r::actor_base_t{cfg}, device_id{cfg.device_id}, announce_url{cfg.announce_url},
      discovery_device_id{std::move(cfg.device_id)}, ssl_pair{*cfg.ssl_pair}, rx_buff_size{cfg.rx_buff_size},
      io_timeout(cfg.io_timeout), debug{cfg.debug}, cluster{cfg.cluster} {
    rx_buff = std::make_shared<rx_buff_t::element_type>(rx_buff_size);
}

void global_discovery_actor_t::configure(r::plugin::plugin_base_t &plugin) noexcept {
    r::actor_base_t::configure(plugin);
    plugin.with_casted<r::plugin::address_maker_plugin_t>([&](auto &p) {
        addr_announce = p.create_address();
        addr_discovery = p.create_address();
        p.set_identity("net.global_discovery", false);
        log = utils::get_logger(identity);
    });
    plugin.with_casted<r::plugin::registry_plugin_t>([&](auto &p) {
        p.discover_name(names::http11_gda, http_client, true).link(true);
        p.discover_name(names::coordinator, coordinator, false).link(false).callback([&](auto phase, auto &ee) {
            if (!ee && phase == r::plugin::registry_plugin_t::phase_t::linking) {
                auto p = get_plugin(r::plugin::starter_plugin_t::class_identity);
                auto plugin = static_cast<r::plugin::starter_plugin_t *>(p);
                plugin->subscribe_actor(&global_discovery_actor_t::on_model_update, coordinator);
            }
        });
    });
    plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
        p.subscribe_actor(&global_discovery_actor_t::on_announce_response, addr_announce);
        p.subscribe_actor(&global_discovery_actor_t::on_discovery_response, addr_discovery);
    });
}

void global_discovery_actor_t::on_start() noexcept {
    LOG_TRACE(log, "on_start (addr = {})", (void *)address.get());
    announce();
    r::actor_base_t::on_start();
}

void global_discovery_actor_t::announce() noexcept {
    LOG_TRACE(log, "announce");

    auto &uris = cluster->get_device()->get_uris();
    if (uris.empty()) {
        return;
    }

    bool has_new = false;
    for (auto &uri : uris) {
        if (!announced_uris.count(uri->buffer())) {
            has_new = true;
            break;
        }
    }
    if (!has_new) {
        return;
    }

    for (auto &uri : uris) {
        announced_uris.emplace(uri->buffer());
    }
    if (log->level() <= spdlog::level::debug) {
        std::string joint_uris;
        for (size_t i = 0; i < uris.size(); ++i) {
            joint_uris += uris[i]->buffer();
            if (i + 1 < uris.size()) {
                joint_uris += ", ";
            }
        }
        LOG_DEBUG(log, "announcing accessibility via {}", joint_uris);
    }

    utils::bytes_t tx_buff;
    auto res = proto::make_announce_request(tx_buff, announce_url, uris);
    if (!res) {
        LOG_TRACE(log, "error making announce request :: {}", res.error().message());
        return do_shutdown(make_error(res.error()));
    }
    make_request(addr_announce, res.value(), std::move(tx_buff));
}

void global_discovery_actor_t::on_announce_response(message::http_response_t &message) noexcept {
    LOG_TRACE(log, "on_announce_response, req = {}", message.payload.request_id());
    resources->release(resource::http);
    http_request.reset();

    auto &ee = message.payload.ee;
    if (ee) {
        LOG_ERROR(log, "announcing error = {}", ee->message());
        auto inner = utils::make_error_code(utils::error_code_t::announce_failed);
        return do_shutdown(make_error(inner, ee));
    }

    auto res = proto::parse_announce(message.payload.res->response);
    if (!res) {
        LOG_WARN(log, "parsing announce error = {}", res.error().message());
        return do_shutdown(make_error(res.error()));
    }

    if (state >= r::state_t::SHUTTING_DOWN)
        return;

    auto reannounce = res.value();
    LOG_DEBUG(log, "will reannounce after {} seconds", reannounce);

    if (resources->has(resource::timer)) {
        cancel_timer(*timer_request);
    }

    auto timeout = pt::seconds(reannounce);
    timer_request = start_timer(timeout, *this, &global_discovery_actor_t::on_timer);
    resources->acquire(resource::timer);

    if (!announced) {
        send<payload::announce_notification_t>(coordinator, get_address());
        announced = true;
    }
}

void global_discovery_actor_t::on_discovery_response(message::http_response_t &message) noexcept {
    using namespace model::diff;
    LOG_TRACE(log, "on_discovery_response");
    resources->release(resource::http);
    http_request.reset();
    auto &custom = message.payload.req->payload.request_payload->custom;
    auto msg = static_cast<model::message::model_update_t *>(custom.get());
    auto diff = static_cast<model::diff::contact::peer_state_t *>(msg->payload.diff.get());
    auto sha256 = diff->peer_id;
    auto it = discovering_devices.find(sha256);
    discovering_devices.erase(it);

    auto &ee = message.payload.ee;
    bool found = false;
    if (ee) {
        LOG_WARN(log, "discovery failed = {}", ee->message());
    } else {
        auto &http_res = message.payload.res->response;
        auto res = proto::parse_contact(http_res);
        if (!res) {
            auto reason = res.error().message();
            auto &body = http_res.body();
            LOG_WARN(log, "parsing discovery error = {}, body({}):\n {}", reason, body.size(), body);
        } else {
            auto device_id = model::device_id_t::from_sha256(sha256).value();
            auto &uris = res.value();
            if (!uris.empty()) {
                LOG_DEBUG(log, "on_discovery_response, found some URIs for {}", device_id);
                found = true;
                if (cluster->get_devices().by_sha256(sha256)) {
                    auto diff = model::diff::cluster_diff_ptr_t{};
                    diff = new contact::update_contact_t(*cluster, device_id, uris);
                    send<model::payload::model_update_t>(coordinator, std::move(diff), this);
                } else {
                    LOG_DEBUG(log, "on_discovery_response, device '{}' is no longer exist", device_id);
                }
            } else {
                LOG_DEBUG(log, "on_discovery_response, no known URIs for {}", device_id);
            }
        }
    }
    if (!found) {
        using state_t = model::device_state_t;
        auto diff = model::diff::contact::peer_state_t::create(*cluster, sha256, nullptr, state_t::offline);
        if (diff) {
            send<model::payload::model_update_t>(coordinator, std::move(diff));
        }
    }
}

void global_discovery_actor_t::on_model_update(model::message::model_update_t &message) noexcept {
    LOG_TRACE(log, "on_model_update");
    auto &diff = *message.payload.diff;
    auto r = diff.visit(*this, &message);
    if (!r) {
        auto ee = make_error(r.assume_error());
        do_shutdown(ee);
    }
}

void global_discovery_actor_t::on_timer(r::request_id_t, bool cancelled) noexcept {
    resources->release(resource::timer);
    announced_uris.clear();
    if (!cancelled && (state == r::state_t::OPERATIONAL)) {
        announce();
    }
}

void global_discovery_actor_t::make_request(const r::address_ptr_t &addr, utils::uri_ptr_t &uri,
                                            utils::bytes_t &&tx_buff, const custom_msg_ptr_t &custom) noexcept {
    auto timeout = r::pt::millisec{io_timeout};
    transport::ssl_junction_t ssl{discovery_device_id, &ssl_pair, true, ""};
    http_request = request_via<payload::http_request_t>(http_client, addr, uri, std::move(tx_buff), rx_buff,
                                                        rx_buff_size, std::move(ssl), debug, custom)
                       .send(timeout);
    resources->acquire(resource::http);
}

void global_discovery_actor_t::shutdown_start() noexcept {
    LOG_TRACE(log, "shutdown_start");
    if (resources->has(resource::http)) {
        send<message::http_cancel_t::payload_t>(http_client, *http_request, get_address());
    }
    send<payload::http_close_connection_t>(http_client);

    if (resources->has(resource::timer)) {
        cancel_timer(*timer_request);
    }
    r::actor_base_t::shutdown_start();
}

auto global_discovery_actor_t::operator()(const model::diff::contact::update_contact_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (diff.self && state == r::state_t::OPERATIONAL) {
        if (resources->has(resource::timer)) {
            cancel_timer(*timer_request);
        }
        announce();
    }
    return diff.visit_next(*this, custom);
}

auto global_discovery_actor_t::operator()(const model::diff::contact::peer_state_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if ((state != r::state_t::OPERATIONAL) || (diff.state != model::device_state_t::discovering)) {
        return diff.visit_next(*this, custom);
    }

    auto sha256 = diff.peer_id;
    if (discovering_devices.count(sha256)) {
        auto device_id = model::device_id_t::from_sha256(sha256).value();
        LOG_TRACE(log, "device '{}' is already discovering, skip", device_id.get_short());
        return diff.visit_next(*this, custom);
    }

    auto peer = cluster->get_devices().by_sha256(sha256);
    if (!peer) {
        LOG_ERROR(log, "no peer device '{}'", peer->device_id());
    } else {
        utils::bytes_t tx_buff;
        auto r = proto::make_discovery_request(tx_buff, announce_url, peer->device_id());
        if (!r) {
            LOG_ERROR(log, "error making discovery request: {}", r.error().message());
            return r.error();
        }

        discovering_devices.emplace(sha256);
        auto msg = reinterpret_cast<model::message::model_update_t *>(custom);
        make_request(addr_discovery, r.value(), std::move(tx_buff), msg);
    }
    return diff.visit_next(*this, custom);
}


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

#pragma once

#include "messages.h"
#include "model/messages.h"
#include "model/diff/cluster_visitor.h"
#include "utils/log.h"
#include "utils/bytes_comparator.hpp"
#include <boost/asio.hpp>
#include <optional>
#include <set>
#include <unordered_set>

namespace syncspirit {
namespace net {

namespace outcome = boost::outcome_v2;

struct global_discovery_actor_config_t : r::actor_config_t {
    utils::uri_ptr_t announce_url;
    model::device_id_t device_id;
    const utils::key_pair_t *ssl_pair;
    std::uint32_t rx_buff_size;
    std::uint32_t io_timeout;
    model::cluster_ptr_t cluster;
    bool debug = false;
};

template <typename Actor> struct global_discovery_actor_config_builder_t : r::actor_config_builder_t<Actor> {
    using builder_t = typename Actor::template config_builder_t<Actor>;
    using parent_t = r::actor_config_builder_t<Actor>;
    using parent_t::parent_t;

    builder_t &&announce_url(const utils::uri_ptr_t &value) && noexcept {
        parent_t::config.announce_url = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&ssl_pair(const utils::key_pair_t *value) && noexcept {
        parent_t::config.ssl_pair = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&rx_buff_size(const std::uint32_t value) && noexcept {
        parent_t::config.rx_buff_size = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&io_timeout(const std::uint32_t value) && noexcept {
        parent_t::config.io_timeout = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&device_id(const model::device_id_t &value) && noexcept {
        parent_t::config.device_id = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&cluster(const model::cluster_ptr_t &value) && noexcept {
        parent_t::config.cluster = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&debug(bool value) && noexcept {
        parent_t::config.debug = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }
};

struct SYNCSPIRIT_API global_discovery_actor_t : public r::actor_base_t, private model::diff::cluster_visitor_t {
    using config_t = global_discovery_actor_config_t;
    template <typename Actor> using config_builder_t = global_discovery_actor_config_builder_t<Actor>;

    explicit global_discovery_actor_t(config_t &cfg);

    void on_start() noexcept override;
    void shutdown_start() noexcept override;
    void configure(r::plugin::plugin_base_t &plugin) noexcept override;

  private:
    using rx_buff_t = payload::http_request_t::rx_buff_ptr_t;
    using discovering_devices_t = std::set<utils::bytes_t, utils::bytes_comparator_t>;
    using uris_t = std::unordered_set<std::string>;
    using custom_msg_ptr_t = r::intrusive_ptr_t<model::message::model_update_t>;

    void announce() noexcept;
    void on_announce_response(message::http_response_t &message) noexcept;
    void on_discovery_response(message::http_response_t &message) noexcept;
    void on_model_update(model::message::model_update_t &message) noexcept;
    void on_timer(r::request_id_t, bool cancelled) noexcept;
    void make_request(const r::address_ptr_t &addr, utils::uri_ptr_t &uri, utils::bytes_t &&tx_buff,
                      const custom_msg_ptr_t &custom = {}) noexcept;

    outcome::result<void> operator()(const model::diff::contact::update_contact_t &, void *custom) noexcept override;
    outcome::result<void> operator()(const model::diff::contact::peer_state_t &, void *custom) noexcept override;

    model::device_id_t device_id;
    utils::logger_t log;
    r::address_ptr_t http_client;
    r::address_ptr_t coordinator;
    utils::uri_ptr_t announce_url;
    model::device_id_t discovery_device_id;
    const utils::key_pair_t &ssl_pair;
    rx_buff_t rx_buff;
    std::uint32_t rx_buff_size;
    std::uint32_t io_timeout;
    bool announced = false;
    bool debug = false;
    r::address_ptr_t addr_announce;  /* for routing */
    r::address_ptr_t addr_discovery; /* for routing */
    std::optional<r::request_id_t> timer_request;
    std::optional<r::request_id_t> http_request;
    discovering_devices_t discovering_devices;
    model::cluster_ptr_t cluster;
    uris_t announced_uris;
};

} // namespace net
} // namespace syncspirit


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

#include "http_actor.h"
#include "utils/error_code.h"
#include "utils/format.hpp"
#include "names.h"

using namespace syncspirit::net;

namespace {
namespace resource {
r::plugin::resource_id_t io = 0;
r::plugin::resource_id_t request_timer = 1;
r::plugin::resource_id_t resolver = 2;
} // namespace resource
} // namespace

http_actor_t::http_actor_t(config_t &config)
    : r::actor_base_t{config}, resolve_timeout(config.resolve_timeout), request_timeout(config.request_timeout),
      registry_name{config.registry_name}, keep_alive{config.keep_alive} {}

void http_actor_t::configure(r::plugin::plugin_base_t &plugin) noexcept {
    r::actor_base_t::configure(plugin);
    plugin.with_casted<r::plugin::address_maker_plugin_t>([&](auto &p) {
        p.set_identity(registry_name, false);
        log = utils::get_logger(identity);
    });
    plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
        p.subscribe_actor(&http_actor_t::on_request);
        p.subscribe_actor(&http_actor_t::on_resolve);
        p.subscribe_actor(&http_actor_t::on_cancel);
        p.subscribe_actor(&http_actor_t::on_close_connection);
    });
    plugin.with_casted<r::plugin::registry_plugin_t>([&](auto &p) {
        p.register_name(registry_name, get_address());
        p.discover_name(names::resolver, resolver, true).link(false);
    });
}

void http_actor_t::on_request(message::http_request_t &req) noexcept {
    LOG_TRACE(log, "on request, url = {}", req.payload.request_payload->url);
    queue.emplace_back(&req);
    process();
}

void http_actor_t::on_cancel(message::http_cancel_t &req) noexcept {
    LOG_TRACE(log, "on_cancel, queue is empty = {}", queue.empty());
    if (queue.empty()) {
        return;
    }

    auto &request_id = req.payload.id;
    if (request_id == queue.front()->payload.id) {
        if (resolve_request) {
            send<message::resolve_cancel_t::payload_t>(resolver, *resolve_request, get_address());
        } else {
            cancel_io();
        }
    } else {
        auto it = queue.begin();
        ++it;
        for (; it != queue.end(); ++it) {
            auto &http_req = **it;
            if (http_req.payload.id == request_id) {
                auto ec = r::make_error_code(r::error_code_t::cancelled);
                reply_with_error(http_req, make_error(ec));
                queue.erase(it);
                return;
            }
        }
    }
}

void http_actor_t::process() noexcept {
    if (stop_io) {
        auto ec = utils::make_error_code(utils::error_code_t::service_not_available);
        for (auto req : queue) {
            reply_with_error(*req, make_error(ec));
        }
        queue.clear();
        if (resources->has(resource::io)) {
            transport->cancel();
        }
        return;
    }
    auto skip =
        queue.empty() || resources->has(resource::io) || resources->has(resource::request_timer) || resolve_request;
    if (skip)
        return;

    http_response.clear();
    need_response = true;
    response_size = 0;
    auto &url = queue.front()->payload.request_payload->url;

    if (keep_alive && kept_alive) {
        bool reuse = false;
        if (url->host() == resolved_url->host() && url->port() == resolved_url->port()) {
            if (!last_read) {
                reuse = true;
            } else {
                auto now = clock_t::local_time();
                auto deadline = *last_read + request_timeout;
                reuse = deadline > now;
            }
        } else {
            LOG_DEBUG(log, "different endpoint is used: {} vs {}", resolved_url, url);
        }
        if (reuse) {
            LOG_DEBUG(log, "reusing connection");
            spawn_timer();
            write_request();
        } else {
            LOG_TRACE(log, "will use new connection");
            kept_alive = false;
            transport->cancel();
            return process();
        }

    } else {
        auto host = url->host();
        auto port = url->port_number();
        resolve_request = request<payload::address_request_t>(resolver, host, port).send(resolve_timeout);
        resources->acquire(resource::resolver);
    }
}

void http_actor_t::spawn_timer() noexcept {
    assert(!timer_request);
    timer_request = start_timer(request_timeout, *this, &http_actor_t::on_timer);
    resources->acquire(resource::request_timer);
}

void http_actor_t::on_resolve(message::resolve_response_t &res) noexcept {
    resources->release(resource::resolver);
    resolve_request.reset();
    auto &ee = res.payload.ee;

    if (ee) {
        LOG_WARN(log, "on_resolve error: {}", ee->message());
        reply_with_error(*queue.front(), ee);
        queue.pop_front();
        need_response = false;
        return process();
    }

    if (stop_io || queue.empty())
        return process();

    if (state != r::state_t::OPERATIONAL) {
        return;
    }

    auto &payload = queue.front()->payload.request_payload;
    auto &ssl_ctx = payload->ssl_context;
    auto sup = static_cast<ra::supervisor_asio_t *>(supervisor);
    transport::transport_config_t cfg{std::move(ssl_ctx), payload->url, *sup, {}, true};
    transport = transport::initiate_http(cfg);
    if (!transport) {
        auto ec = utils::make_error_code(utils::error_code_t::transport_not_available);
        reply_with_error(*queue.front(), make_error(ec));
        queue.pop_front();
        need_response = false;
        return process();
    }

    auto &ips = res.payload.res->results;
    auto port = res.payload.req->payload.request_payload->port;
    auto endpoints = utils::make_endpoints<tcp, tcp::endpoint>(ips, port);

    transport::connect_fn_t on_connect = [&](const auto &arg) { this->on_connect(arg); };
    transport::error_fn_t on_error = [&](auto arg) { this->on_io_error(arg); };
    transport->async_connect(std::move(endpoints), on_connect, on_error);
    resources->acquire(resource::io);
    spawn_timer();
    resolved_url = payload->url;
}

void http_actor_t::on_connect(const tcp::endpoint &) noexcept {
    LOG_TRACE(log, "on_connect");
    resources->release(resource::io);
    if (!need_response || stop_io) {
        return process();
    }

    if (queue.front()->payload.request_payload->local_ip) {
        sys::error_code ec;
        local_address = transport->local_address(ec);
        if (ec) {
            LOG_WARN(log, "on_connect, get local addr error :: {}", ec.message());
            reply_with_error(*queue.front(), make_error(ec));
            queue.pop_front();
            need_response = false;
            if (resources->has(resource::request_timer)) {
                cancel_timer(*timer_request);
            }
            return process();
        }
    }

    transport::handshake_fn_t handshake_fn([&](auto &&...args) { on_handshake(args...); });
    transport::error_fn_t error_fn([&](auto arg) { on_handshake_error(arg); });
    transport->async_handshake(handshake_fn, error_fn);
    resources->acquire(resource::io);
}

void http_actor_t::write_request() noexcept {
    auto &payload = *queue.front()->payload.request_payload;
    auto &url = payload.url;
    auto &data = payload.data;
    LOG_TRACE(log, "sending {} bytes to {} ", data.size(), url);
    auto buff = asio::buffer(data.data(), data.size());
    if (payload.debug) {
        std::string_view write_data{(const char *)buff.data(), data.size()};
        LOG_DEBUG(log, "request ({}):\n{}", data.size(), write_data);
    }
    transport::io_fn_t on_write = [&](auto arg) { this->on_request_sent(arg); };
    transport::error_fn_t on_error = [&](auto arg) { this->on_io_error(arg); };
    transport->async_send(buff, on_write, on_error);
    resources->acquire(resource::io);
}

void http_actor_t::on_request_sent(std::size_t /* bytes */) noexcept {
    resources->release(resource::io);
    if (!need_response || stop_io) {
        return process();
    }

    auto &payload = *queue.front()->payload.request_payload;
    auto &rx_buff = payload.rx_buff;
    rx_buff->prepare(payload.rx_buff_size);
    transport::io_fn_t on_read = [&](auto arg) { this->on_request_read(arg); };
    transport::error_fn_t on_error = [&](auto arg) { this->on_io_error(arg); };
    transport->async_read(*rx_buff, http_response, on_read, on_error);
    resources->acquire(resource::io);
}

void http_actor_t::on_request_read(std::size_t bytes) noexcept {
    resources->release(resource::io);
    response_size = bytes;

    auto &req = *queue.front();
    if (req.payload.request_payload->debug) {
        auto &body = http_response.body();
        LOG_DEBUG(log, "response (bytes: {}, code: {}):\n{}", bytes, http_response.result_int(), body);
    }
    if (keep_alive && http_response.keep_alive()) {
        kept_alive = true;
        last_read = clock_t::local_time();
    } else {
        kept_alive = false;
        transport.reset();
    }

    if (resources->has(resource::request_timer)) {
        cancel_timer(*timer_request);
    }

    reply_to(req, std::move(http_response), response_size, std::move(local_address));
    queue.pop_front();
    need_response = false;

    process();
}

void http_actor_t::on_io_error(const sys::error_code &ec) noexcept {
    resources->release(resource::io);
    kept_alive = false;
    if (ec != asio::error::operation_aborted) {
        LOG_DEBUG(log, "on_io_error :: {}", ec.message());
    }
    cancel_io();
    if (!need_response || stop_io) {
        return process();
    }

    reply_with_error(*queue.front(), make_error(ec));
    queue.pop_front();
    need_response = false;
}

void http_actor_t::on_handshake(bool, utils::x509_t &, const tcp::endpoint &, const model::device_id_t *) noexcept {
    resources->release(resource::io);
    if (!need_response || stop_io) {
        return process();
    }
    if (state == r::state_t::OPERATIONAL) {
        write_request();
    }
}

void http_actor_t::on_handshake_error(sys::error_code ec) noexcept {
    resources->release(resource::io);
    if (ec != asio::error::operation_aborted) {
        LOG_WARN(log, "on_handshake_error :: {}", ec.message());
    }
    if (!need_response || stop_io) {
        return process();
    }
    reply_with_error(*queue.front(), make_error(ec));
    queue.pop_front();
    need_response = false;
    cancel_io();
    process();
}

void http_actor_t::on_timer(r::request_id_t, bool cancelled) noexcept {
    resources->release(resource::request_timer);
    timer_request.reset();

    bool do_cancel_io = false;
    if (!cancelled) {
        auto ec = r::make_error_code(r::error_code_t::request_timeout);
        reply_with_error(*queue.front(), make_error(ec));
        queue.pop_front();
        need_response = false;
        if (!kept_alive) {
            do_cancel_io = true;
        }
    } else {
        do_cancel_io = true;
    }

    if (do_cancel_io) {
        cancel_io();
    }
}

void http_actor_t::on_start() noexcept {
    LOG_TRACE(log, "on_start");
    r::actor_base_t::on_start();
}

void http_actor_t::shutdown_finish() noexcept {
    LOG_TRACE(log, "shutdown_finish");
    r::actor_base_t::shutdown_finish();
    transport.reset();
}

void http_actor_t::on_close_connection(message::http_close_connection_t &) noexcept {
    LOG_TRACE(log, "on_close_connection");
    stop_io = true;
    cancel_io();
}

void http_actor_t::cancel_io() noexcept {
    if (resources->has(resource::io)) {
        transport->cancel();
    }
    if (resources->has(resource::resolver)) {
        send<message::resolve_cancel_t::payload_t>(resolver, *resolve_request, get_address());
    }
    if (resources->has(resource::request_timer)) {
        cancel_timer(*timer_request);
    }
}


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

#pragma once

#include "transport/http.h"
#include "utils/log.h"
#include "messages.h"
#include <boost/asio.hpp>
#include <rotor.hpp>
#include <optional>

namespace syncspirit {
namespace net {

struct http_actor_config_t : public r::actor_config_t {
    using r::actor_config_t::actor_config_t;
    r::pt::time_duration resolve_timeout;
    r::pt::time_duration request_timeout;
    std::string registry_name;
    bool keep_alive;
};

template <typename Actor> struct http_actor_config_builder_t : r::actor_config_builder_t<Actor> {
    using builder_t = typename Actor::template config_builder_t<Actor>;
    using parent_t = r::actor_config_builder_t<Actor>;
    using parent_t::parent_t;

    builder_t &&resolve_timeout(const pt::time_duration &value) && noexcept {
        parent_t::config.resolve_timeout = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&request_timeout(const pt::time_duration &value) && noexcept {
        parent_t::config.request_timeout = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&registry_name(const std::string &value) && noexcept {
        parent_t::config.registry_name = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&keep_alive(bool value = true) && noexcept {
        parent_t::config.keep_alive = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }
};

struct SYNCSPIRIT_API http_actor_t : public r::actor_base_t {
    using request_ptr_t = r::intrusive_ptr_t<message::http_request_t>;
    using socket_ptr_t = std::unique_ptr<tcp::socket>;
    using secure_socket_t = std::unique_ptr<ssl::stream<tcp_socket_t>>;

    using resolve_it_t = payload::address_response_t::resolve_results_t::iterator;

    using config_t = http_actor_config_t;
    template <typename Actor> using config_builder_t = http_actor_config_builder_t<Actor>;

    explicit http_actor_t(config_t &config);

    void configure(r::plugin::plugin_base_t &plugin) noexcept override;
    void on_start() noexcept override;
    void shutdown_finish() noexcept override;

  private:
    using queue_t = std::list<request_ptr_t>;
    using request_option_t = std::optional<r::request_id_t>;
    using clock_t = r::pt::microsec_clock;

    void process() noexcept;
    void spawn_timer() noexcept;

    void on_request(message::http_request_t &req) noexcept;
    void on_cancel(message::http_cancel_t &req) noexcept;
    void on_resolve(message::resolve_response_t &res) noexcept;
    void on_close_connection(message::http_close_connection_t &) noexcept;

    void on_connect(const tcp::endpoint &) noexcept;
    void on_request_sent(std::size_t /* bytes */) noexcept;
    void on_request_read(std::size_t bytes) noexcept;
    void on_timer(r::request_id_t, bool cancelled) noexcept;
    void on_io_error(const sys::error_code &ec) noexcept;
    void on_shutdown_timer(r::request_id_t, bool cancelled) noexcept;
    void on_handshake(bool valid_peer, utils::x509_t &, const tcp::endpoint &, const model::device_id_t *) noexcept;
    void on_handshake_error(sys::error_code ec) noexcept;
    void cancel_io() noexcept;
    void write_request() noexcept;
    void start_shutdown_timer() noexcept;

    utils::logger_t log;
    pt::time_duration resolve_timeout;
    pt::time_duration request_timeout;
    std::string registry_name;
    bool keep_alive;
    bool kept_alive = false;
    r::address_ptr_t resolver;
    queue_t queue;
    bool need_response = false;
    bool stop_io = false;
    transport::http_sp_t transport;
    std::optional<asio::ip::address> local_address;
    http::request<http::empty_body> http_request;
    http::response<http::string_body> http_response;
    size_t response_size = 0;
    utils::uri_ptr_t resolved_url;
    request_option_t resolve_request;
    request_option_t timer_request;
    request_option_t shutdown_request;
    std::optional<r::pt::ptime> last_read;
};

} // namespace net
} // namespace syncspirit


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

#include "initiator_actor.h"
#include "constants.h"
#include "names.h"
#include "model/messages.h"
#include "proto/relay_support.h"
#include "utils/error_code.h"
#include "utils/format.hpp"
#include "model/diff/contact/peer_state.h"
#include <algorithm>
#include <spdlog/fmt/bin_to_hex.h>
#include <fmt/core.h>
#include <ares.h>

using namespace syncspirit::net;

namespace {
namespace resource {
r::plugin::resource_id_t initializing = 0;
r::plugin::resource_id_t resolving = 1;
r::plugin::resource_id_t connect = 2;
r::plugin::resource_id_t handshake = 3;
r::plugin::resource_id_t read = 4;
r::plugin::resource_id_t write = 5;
} // namespace resource
} // namespace

static constexpr size_t BUFF_SZ = 256;

initiator_actor_t::initiator_actor_t(config_t &cfg)
    : r::actor_base_t{cfg}, peer_device_id{cfg.peer_device_id}, relay_key(std::move(cfg.relay_session)),
      ssl_pair{*cfg.ssl_pair}, sock(std::move(cfg.sock)), cluster{std::move(cfg.cluster)}, sink(std::move(cfg.sink)),
      custom(std::move(cfg.custom)), router{*cfg.router}, alpn(cfg.alpn) {
    auto tmp_identity = "net.init/unknown";
    auto log = utils::get_logger(tmp_identity);
    for (auto &uri : cfg.uris) {
        if (uri->scheme() != "tcp" && uri->scheme() != "relay") {
            LOG_DEBUG(log, "unsupported scheme in the url '{}'", uri);
        } else {
            if (uri->scheme() == "relay" && !cfg.relay_enabled) {
                LOG_DEBUG(log, "'relay' is not enabled, skipping '{}'", uri);
            } else {
                uris.emplace_back(std::move(uri));
            }
        }
    }
    auto comparator = [](const utils::uri_ptr_t &a, const utils::uri_ptr_t &b) noexcept -> bool {
        if (a->scheme() == b->scheme()) {
            auto x = a->buffer();
            auto y = b->buffer();
            return std::lexicographical_compare(x.begin(), x.end(), y.begin(), y.end());
        }
        if (a->scheme() == "relay") {
            return true;
        }
        return false;
    };
    std::sort(uris.begin(), uris.end(), comparator);

    if (sock.has_value()) {
        role = role_t::passive;
    } else {
        if (!relay_key.empty()) {
            role = role_t::relay_passive;
        } else {
            role = role_t::active;
        }
    }
}

void initiator_actor_t::configure(r::plugin::plugin_base_t &plugin) noexcept {
    r::actor_base_t::configure(plugin);

    plugin.with_casted<r::plugin::address_maker_plugin_t>([&](auto &p) {
        std::string value;
        switch (role) {
        case role_t::active:
            value = fmt::format("init/active:{}", peer_device_id.get_short());
            break;
        case role_t::passive: {
            auto ep = sock.value().remote_endpoint();
            value = fmt::format("init/passive:{}", ep);
            break;
        }
        case role_t::relay_passive: {
            value = fmt::format("init/relay-passive:{}", peer_device_id.get_short());
            break;
        }
        }
        p.set_identity("net." + value, false);
        log = utils::get_logger(identity);
    });
    plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
        p.subscribe_actor(&initiator_actor_t::on_resolve);
        resources->acquire(resource::initializing);
        if (role == role_t::active) {
            initiate_active();
        } else if (role == role_t::passive) {
            initiate_passive();
        } else if (role == role_t::relay_passive) {
            initiate_relay_passive();
        }
    });
    plugin.with_casted<r::plugin::registry_plugin_t>([&](auto &p) {
        p.discover_name(names::resolver, resolver).link(false);
        p.discover_name(names::coordinator, coordinator).link(false).callback([&](auto phase, auto &ee) {
            if (!ee && phase == r::plugin::registry_plugin_t::phase_t::linking && role == role_t::active) {
                auto diff = model::diff::cluster_diff_ptr_t();
                auto state = model::device_state_t::connecting;
                auto sha256 = peer_device_id.get_sha256();
                auto peer = cluster && cluster->get_devices().by_sha256(sha256);
                if (peer) {
                    diff = model::diff::contact::peer_state_t::create(*cluster, sha256, nullptr, state);
                    if (diff) {
                        send<model::payload::model_update_t>(coordinator, std::move(diff));
                    }
                }
            }
        });
    });
}

void initiator_actor_t::initiate_active() noexcept {
    if (state > r::state_t::OPERATIONAL) {
        return;
    }

    while (uri_idx < uris.size()) {
        auto &uri = uris[uri_idx++];
        if (uri->scheme() == "tcp") {
            initiate_active_tls(uri);
        } else if (uri->scheme() == "relay") {
            initiate_active_relay(uri);
        } else {
            continue;
        }
        return;
    }

    LOG_TRACE(log, "try_next_uri, no way to connect found, shut down");
    auto ec = utils::make_error_code(utils::error_code_t::connection_impossible);
    do_shutdown(make_error(ec));
}

void initiator_actor_t::initiate_passive() noexcept {
    if (state > r::state_t::OPERATIONAL) {
        return;
    }

    auto sup = static_cast<ra::supervisor_asio_t *>(&router);
    transport = transport::initiate_tls_passive(*sup, ssl_pair, std::move(sock.value()), alpn);
    initiate_handshake();
}

void initiator_actor_t::initiate_relay_passive() noexcept {
    if (state > r::state_t::OPERATIONAL) {
        return;
    }

    auto sup = static_cast<ra::supervisor_asio_t *>(&router);
    auto &uri = uris.at(0);
    transport::transport_config_t cfg{{}, uri, *sup, {}, true};
    transport = transport::initiate_stream(cfg);
    assert(transport);
    resolve(uri);
}

void initiator_actor_t::on_start() noexcept {
    r::actor_base_t::on_start();
    LOG_TRACE(log, "on_start, alpn = {}", alpn);
    std::string proto;
    if (active_uri) {
        proto = active_uri->scheme();
    } else {
        if (role == role_t::relay_passive) {
            proto = "relay";
        } else if (role == role_t::passive) {
            proto = "tcp";
        }
    }
    send<payload::peer_connected_t>(sink, std::move(transport), peer_device_id, remote_endpoint, std::move(proto),
                                    std::move(custom));
    success = true;
    do_shutdown();
}

void initiator_actor_t::shutdown_start() noexcept {
    LOG_TRACE(log, "shutdown_start");
    if (resources->has(resource::initializing)) {
        resources->release(resource::initializing);
    }
    bool cancel_transport = resources->has(resource::connect) || resources->has(resource::handshake) ||
                            resources->has(resource::read) || resources->has(resource::write);
    if (cancel_transport) {
        transport->cancel();
    }
    r::actor_base_t::shutdown_start();
}

void initiator_actor_t::shutdown_finish() noexcept {
    LOG_TRACE(log, "shutdown_finish");
    bool notify_offline = role == role_t::active && !success && cluster;
    if (notify_offline) {
        auto state = model::device_state_t::offline;
        auto sha256 = peer_device_id.get_sha256();
        if (auto peer = cluster->get_devices().by_sha256(sha256); peer) {
            auto diff = model::diff::contact::peer_state_t::create(*cluster, sha256, nullptr, state);
            if (diff) {
                send<model::payload::model_update_t>(coordinator, std::move(diff));
            }
        }
    }
    r::actor_base_t::shutdown_finish();
}

void initiator_actor_t::resolve(const utils::uri_ptr_t &uri) noexcept {
    LOG_DEBUG(log, "resolving {} (transport = {})", uri, (void *)transport.get());
    pt::time_duration resolve_timeout = init_timeout / 2;
    auto host = uri->host();
    auto port = uri->port_number();
    request<payload::address_request_t>(resolver, host, port).send(resolve_timeout);
    resources->acquire(resource::resolving);
}

void initiator_actor_t::initiate_active_tls(const utils::uri_ptr_t &uri) noexcept {
    LOG_DEBUG(log, "trying '{}' as active tls, alpn = {}", uri, alpn);
    auto sup = static_cast<ra::supervisor_asio_t *>(&router);
    transport = transport::initiate_tls_active(*sup, ssl_pair, peer_device_id, uri, false, alpn);
    active_uri = uri;
    relaying = false;
    resolve(uri);
}

void initiator_actor_t::initiate_active_relay(const utils::uri_ptr_t &uri) noexcept {
    LOG_TRACE(log, "trying '{}' as active relay", uri);
    auto relay_device = proto::relay::parse_device(uri);
    if (!relay_device) {
        LOG_WARN(log, "relay url '{}' does not contains valid device_id", uri);
        return initiate_active();
    }
    active_uri = uri;
    relaying = true;
    auto sup = static_cast<ra::supervisor_asio_t *>(&router);
    transport = transport::initiate_tls_active(*sup, ssl_pair, relay_device.value(), active_uri);
    resolve(uri);
}

void initiator_actor_t::on_resolve(message::resolve_response_t &res) noexcept {
    LOG_TRACE(log, "on_resolve");
    resources->release(resource::resolving);
    if (state > r::state_t::OPERATIONAL) {
        return;
    }

    auto &ee = res.payload.ee;
    if (ee) {
        LOG_WARN(log, "on_resolve error : {}", ee->message());
        if (role == role_t::active) {
            return initiate_active();
        } else {
            return do_shutdown(ee);
        }
    }

    auto &ips = res.payload.res->results;
    auto port = res.payload.req->payload.request_payload->port;
    auto endpoints = utils::make_endpoints<tcp, tcp::endpoint>(ips, port);

    transport::connect_fn_t on_connect = [&](const auto &arg) { this->on_connect(arg); };
    transport::error_fn_t on_error = [&](auto arg) { this->on_io_error(arg, resource::connect); };
    transport->async_connect(std::move(endpoints), on_connect, on_error);
    resources->acquire(resource::connect);
}

void initiator_actor_t::on_io_error(const sys::error_code &ec, r::plugin::resource_id_t resource) noexcept {
    LOG_TRACE(log, "on_io_error: {}", ec.message());
    resources->release(resource);
    if (ec != asio::error::operation_aborted) {
        LOG_WARN(log, "on_io_error: {}", ec.message());
    }
    if (state < r::state_t::SHUTTING_DOWN) {
        if (!connected && role == role_t::active) {
            initiate_active();
        } else {
            connected = false;
            LOG_DEBUG(log, "on_io_error, initiating shutdown...");
            do_shutdown(make_error(ec));
        }
    }
}

void initiator_actor_t::on_connect(const tcp::endpoint &) noexcept {
    LOG_TRACE(log, "on_connect, device_id = {}, transport = {}", peer_device_id.get_short(), (void *)transport.get());
    resources->release(resource::connect);
    // auto do_handshake = role == role_t::active;
    auto do_handshake =
        (role == role_t::active) &&
        (active_uri && ((((active_uri->scheme() == "relay") && relaying)) || (active_uri->scheme() == "tcp")));
    if (do_handshake) {
        initiate_handshake();
    } else {
        join_session();
    }
}

void initiator_actor_t::initiate_handshake() noexcept {
    if (state > r::state_t::OPERATIONAL) {
        return;
    }

    LOG_TRACE(log, "initializing handshake");
    connected = true;
    transport::handshake_fn_t handshake_fn([&](auto &&...args) { on_handshake(args...); });
    transport::error_fn_t error_fn([&](auto arg) { on_io_error(arg, resource::handshake); });
    resources->acquire(resource::handshake);
    transport->async_handshake(handshake_fn, error_fn);
}

void initiator_actor_t::join_session() noexcept {
    if (state > r::state_t::OPERATIONAL) {
        return;
    }

    transport::error_fn_t read_err_fn([&](auto arg) { on_io_error(arg, resource::read); });
    transport::io_fn_t read_fn = [this](size_t bytes) { on_read_relay(bytes); };
    rx_buff.resize(BUFF_SZ);
    transport->async_recv(asio::buffer(rx_buff), read_fn, read_err_fn);
    resources->acquire(resource::read);

    LOG_TRACE(log, "join_session");
    auto msg = proto::relay::join_session_request_t{std::move(relay_key)};
    proto::relay::serialize(msg, relay_tx);
    transport::error_fn_t write_err_fn([&](auto arg) { on_io_error(arg, resource::write); });
    transport::io_fn_t write_fn = [this](size_t bytes) { on_write(bytes); };
    transport->async_send(asio::buffer(relay_tx), write_fn, write_err_fn);
    resources->acquire(resource::write);
}

void initiator_actor_t::on_handshake(bool valid_peer, utils::x509_t &cert, const tcp::endpoint &peer_endpoint,
                                     const model::device_id_t *peer_device) noexcept {
    resources->release(resource::handshake);
    if (!peer_device) {
        LOG_WARN(log, "on_handshake,  missing peer device id");
        auto ec = utils::make_error_code(utils::error_code_t::missing_device_id);
        return do_shutdown(make_error(ec));
    }

    auto cert_name = utils::get_common_name(cert);
    if (!cert_name) {
        LOG_WARN(log, "on_handshake, can't get certificate name: {}", cert_name.error().message());
        auto ec = utils::make_error_code(utils::error_code_t::missing_cn);
        return do_shutdown(make_error(ec));
    }
    LOG_TRACE(log, "on_handshake, valid = {}, issued by {}", valid_peer, cert_name.value());
    if (relaying) {
        request_relay_connection();
    } else {
        peer_device_id = *peer_device;
        remote_endpoint = peer_endpoint;
        if (state <= r::state_t::OPERATIONAL) {
            resources->release(resource::initializing);
        }
    }
}

void initiator_actor_t::on_write(size_t bytes) noexcept {
    LOG_TRACE(log, "on_write, {} bytes", bytes);
    resources->release(resource::write);
}

void initiator_actor_t::on_read_relay(size_t bytes) noexcept {
    LOG_TRACE(log, "on_read_relay, {} bytes", bytes);
    resources->release(resource::read);
    if (state > r::state_t::OPERATIONAL) {
        return;
    }

    auto buff = utils::bytes_view_t(rx_buff.data(), bytes);
    auto r = proto::relay::parse(buff);
    auto wrapped = std::get_if<proto::relay::wrapped_message_t>(&r);
    if (!wrapped) {
        LOG_WARN(log, "unexpected incoming relay data: {}", spdlog::to_hex(buff.begin(), buff.end()));
        auto ec = utils::make_error_code(utils::error_code_t::relay_failure);
        return do_shutdown(make_error(ec));
    }
    auto reply = std::get_if<proto::relay::response_t>(&wrapped->message);
    if (!reply) {
        LOG_WARN(log, "unexpected relay message: {}", spdlog::to_hex(buff.begin(), buff.end()));
        auto ec = utils::make_error_code(utils::error_code_t::relay_failure);
        return do_shutdown(make_error(ec));
    }
    if (reply->code) {
        LOG_WARN(log, "relay join failure({}): {}", reply->code, reply->details);
        auto ec = utils::make_error_code(utils::error_code_t::relay_failure);
        return do_shutdown(make_error(ec));
    }
    auto &upgradeable = dynamic_cast<transport::upgradeable_stream_base_t &>(*transport.get());
    auto ssl = transport::ssl_junction_t{peer_device_id, &ssl_pair, true, constants::protocol_name};
    auto active = role == role_t::active;
    transport = upgradeable.upgrade(ssl, active);
    initiate_handshake();
}

void initiator_actor_t::request_relay_connection() noexcept {
    if (state > r::state_t::OPERATIONAL) {
        return;
    }

    transport::error_fn_t read_err_fn([&](auto arg) { on_io_error(arg, resource::read); });
    transport::io_fn_t read_fn = [this](size_t bytes) { on_read_relay_active(bytes); };
    rx_buff.resize(BUFF_SZ);
    transport->async_recv(asio::buffer(rx_buff), read_fn, read_err_fn);
    resources->acquire(resource::read);

    auto sha256 = peer_device_id.get_sha256();
    auto msg = proto::relay::connect_request_t(utils::bytes_t(sha256.begin(), sha256.end()));
    proto::relay::serialize(msg, relay_tx);
    transport::error_fn_t write_err_fn([&](auto arg) { on_io_error(arg, resource::write); });
    transport::io_fn_t write_fn = [this](size_t bytes) { on_write(bytes); };
    transport->async_send(asio::buffer(relay_tx), write_fn, write_err_fn);
    resources->acquire(resource::write);
}

void initiator_actor_t::on_read_relay_active(size_t bytes) noexcept {
    LOG_TRACE(log, "on_read_relay_active, {} bytes", bytes);
    resources->release(resource::read);
    auto buff = utils::bytes_view_t(rx_buff.data(), bytes);
    auto r = proto::relay::parse(buff);
    auto wrapped = std::get_if<proto::relay::wrapped_message_t>(&r);
    if (!wrapped) {
        LOG_WARN(log, "unexpected incoming relay data: {}", spdlog::to_hex(buff.begin(), buff.end()));
        auto ec = utils::make_error_code(utils::error_code_t::relay_failure);
        return do_shutdown(make_error(ec));
    }
    auto inv = std::get_if<proto::relay::session_invitation_t>(&wrapped->message);
    if (!inv) {
        auto reply = std::get_if<proto::relay::response_t>(&wrapped->message);
        if (reply) {
            LOG_DEBUG(log, "reply (expected session invitation) ({}) : {}", reply->code, reply->details);
        } else {
            LOG_WARN(log, "unexpected relay message: {}", spdlog::to_hex(buff.begin(), buff.end()));
        }
        return initiate_active();
    }
    auto &peer = inv->from;
    if (peer != peer_device_id.get_sha256()) {
        LOG_WARN(log, "unexpected peer device: {}", spdlog::to_hex(peer.begin(), peer.end()));
        return initiate_active();
    }
    auto &addr = inv->address;
    char addr_buff[INET_ADDRSTRLEN + 1] = {0};
    using string_t = decltype(active_uri->host());
    auto host = string_t{};
    auto ip = std::string_view();
    if (addr.has_value()) {
        auto raw_value = addr.value().to_uint();
        ares_inet_ntop(AF_INET, &raw_value, addr_buff, INET_ADDRSTRLEN);
        ip = addr_buff;
    } else {
        host = active_uri->host();
        ip = host;
    }

    relay_key = inv->key;
    auto uri_str = fmt::format("tcp://{}:{}", ip, inv->port);
    LOG_DEBUG(log, "going to connect to {}, using key: {}", uri_str,
              spdlog::to_hex(relay_key.begin(), relay_key.end()));
    auto uri = utils::parse(uri_str);
    relaying = false;

    auto sup = static_cast<ra::supervisor_asio_t *>(&router);
    transport::transport_config_t cfg{{}, uri, *sup, {}, true};
    transport = transport::initiate_stream(cfg);
    resolve(uri);
}