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

#include "scan_actor.h"
#include "model/diff/advance/local_update.h"
#include "model/diff/modify/lock_file.h"
#include "model/diff/local/blocks_availability.h"
#include "model/diff/local/io_failure.h"
#include "model/diff/local/file_availability.h"
#include "model/diff/local/scan_finish.h"
#include "model/diff/local/scan_start.h"
#include "presentation/cluster_file_presence.h"
#include "presentation/folder_entity.h"
#include "presentation/presence.h"
#include "net/names.h"

#include <algorithm>
#include <memory_resource>
#include <list>
#include <iterator>

namespace sys = boost::system;
using namespace syncspirit::fs;

template <class> inline constexpr bool always_false_v = false;

scan_actor_t::scan_actor_t(config_t &cfg)
    : r::actor_base_t{cfg}, cluster{cfg.cluster}, sequencer{cfg.sequencer}, fs_config{cfg.fs_config},
      requested_hashes_limit{cfg.requested_hashes_limit} {
    assert(sequencer);
}

void scan_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::names::fs_scanner, false);
        log = utils::get_logger(identity);
        new_files = p.create_address();
    });
    plugin.with_casted<r::plugin::registry_plugin_t>([&](auto &p) {
        p.register_name(net::names::fs_scanner, address);
        p.discover_name(net::names::hasher_proxy, hasher_proxy, true).link();
        p.discover_name(net::names::coordinator, coordinator, true).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(&scan_actor_t::on_model_update, coordinator);
            }
        });
    });
    plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
        p.subscribe_actor(&scan_actor_t::on_scan);
        p.subscribe_actor(&scan_actor_t::on_hash);
        p.subscribe_actor(&scan_actor_t::on_hash_new, new_files);
        p.subscribe_actor(&scan_actor_t::on_rehash);
        p.subscribe_actor(&scan_actor_t::on_hash_anew);
    });
}

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

void scan_actor_t::shutdown_finish() noexcept {
    LOG_TRACE(log, "shutdown_finish");
    get_supervisor().shutdown();
    r::actor_base_t::shutdown_finish();
}

auto scan_actor_t::operator()(const model::diff::local::scan_start_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    LOG_DEBUG(log, "initiating scan of {}", diff.folder_id);
    auto task = scan_task_ptr_t(new scan_task_t(cluster, diff.folder_id, fs_config));
    send<payload::scan_progress_t>(address, std::move(task));
    return diff.visit_next(*this, custom);
}

void scan_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, nullptr);
    if (!r) {
        auto ee = make_error(r.assume_error());
        do_shutdown(ee);
    }
}

void scan_actor_t::on_scan(message::scan_progress_t &message) noexcept {
    auto &task = message.payload.task;
    auto guard = task->guard(*this, coordinator);
    auto folder_id = task->get_folder_id();
    LOG_TRACE(log, "on_scan, folder = {}", folder_id);

    auto folder = cluster->get_folders().by_id(folder_id);
    bool stop_processing = false;
    bool completed = false;
    if (folder->is_synchronizing() || folder->is_suspended()) {
        LOG_DEBUG(log, "cancelling folder '{}' scanning", folder_id);
        stop_processing = true;
        completed = true;
    } else {
        auto r = task->advance();
        std::visit(
            [&](auto &&r) {
                using T = std::decay_t<decltype(r)>;
                if constexpr (std::is_same_v<T, bool>) {
                    stop_processing = !r;
                    if (stop_processing) {
                        completed = true;
                    }
                } else if constexpr (std::is_same_v<T, scan_errors_t>) {
                    task->push(new model::diff::local::io_failure_t(std::move(r)));
                } else if constexpr (std::is_same_v<T, unchanged_meta_t>) {
                    task->push(new model::diff::local::file_availability_t(r.file));
                } else if constexpr (std::is_same_v<T, removed_t>) {
                    auto &file = *r.file;
                    LOG_DEBUG(log, "locally removed {}", file.get_full_name());
                    auto folder = file.get_folder_info()->get_folder();
                    auto folder_id = std::string(folder->get_id());
                    auto fi = file.as_proto(false);
                    proto::set_deleted(fi, true);
                    task->push(new model::diff::advance::local_update_t(*cluster, *sequencer, std::move(fi),
                                                                        std::move(folder_id)));
                } else if constexpr (std::is_same_v<T, changed_meta_t>) {
                    auto &file = *r.file;
                    auto &metadata = r.metadata;
                    auto errs = initiate_hash(task, file.get_path(), metadata);
                    if (errs.empty()) {
                        stop_processing = true;
                    } else {
                        task->push(new model::diff::local::io_failure_t(std::move(errs)));
                    }
                } else if constexpr (std::is_same_v<T, unknown_file_t>) {
                    auto errs = initiate_hash(task, r.path, r.metadata);
                    if (errs.empty()) {
                        stop_processing = true;
                    } else {
                        task->push(new model::diff::local::io_failure_t(std::move(errs)));
                    }
                } else if constexpr (std::is_same_v<T, incomplete_t>) {
                    auto &f = r.file;
                    stop_processing = true;
                    send<payload::rehash_needed_t>(address, task, std::move(f), std::move(r.opened_file));
                } else if constexpr (std::is_same_v<T, incomplete_removed_t>) {
                    auto &file = *r.file;
                    if (file.is_locked()) {
                        task->push(new model::diff::modify::lock_file_t(file, false));
                    }
                } else if constexpr (std::is_same_v<T, orphaned_removed_t>) {
                    // NO-OP
                } else if constexpr (std::is_same_v<T, file_error_t>) {
#if 0
                    auto diff = model::diff::cluster_diff_ptr_t{};
                    diff = new model::diff::modify::lock_file_t(*r.file, false);
                    send<model::payload::model_update_t>(coordinator, /std::move(diff), this);
                    auto path = make_temporal(r.file->get_path());
                    scan_errors_t errors{scan_error_t{path, r.ec}};
                    send<model::payload::io_error_t>(coordinator, std::move(errors));
#endif
                    std::abort();
                } else {
                    static_assert(always_false_v<T>, "non-exhaustive visitor!");
                }
            },
            r);
    }

    if (!stop_processing) {
        guard.send_progress();
    } else if (completed) {
        LOG_DEBUG(log, "completed scanning of {}", folder_id);
        post_scan(*task);
        auto now = clock_t::local_time();
        task->push(new model::diff::local::scan_finish_t(folder_id, now));
        guard.send_by_force();
    }
}

auto scan_actor_t::initiate_hash(scan_task_ptr_t task, const bfs::path &path, proto::FileInfo &metadata) noexcept
    -> scan_errors_t {
    file_ptr_t file;
    LOG_DEBUG(log, "will try to initiate hashing of {}", path.string());
    if (proto::get_type(metadata) == proto::FileInfoType::FILE) {
        auto opt = file_t::open_read(path);
        if (!opt) {
            auto &ec = opt.assume_error();
            LOG_WARN(log, "error opening file {}: {}", path.string(), ec.message());
            return scan_errors_t{scan_error_t{path, ec}};
        }
        file = new file_t(std::move(opt.value()));
    }
    send<payload::hash_anew_t>(address, std::move(task), std::move(metadata), std::move(file));
    return {};
}

void scan_actor_t::on_rehash(message::rehash_needed_t &message) noexcept {
    LOG_TRACE(log, "on_rehash");
    auto initial = requested_hashes;
    hash_next(message, address);
    if (initial == requested_hashes) {
        send<payload::scan_progress_t>(address, message.payload.get_task());
    }
}

void scan_actor_t::on_hash_anew(message::hash_anew_t &message) noexcept {
    LOG_TRACE(log, "on_hash_anew");
    auto initial = requested_hashes;
    hash_next(message, new_files);
    // file w/o context
    if (initial == requested_hashes) {
        auto &p = message.payload;
        commit_new_file(p);
        send<payload::scan_progress_t>(address, p.get_task());
    }
}

template <typename Message>
void scan_actor_t::hash_next(Message &message, const r::address_ptr_t &reply_addr) noexcept {
    auto &info = message.payload;
    if (info.has_more_chunks()) {
        auto condition = [&]() { return requested_hashes < requested_hashes_limit && info.has_more_chunks(); };
        while (condition()) {
            auto opt = info.read();
            if (!opt) {
                auto &task = *info.get_task().get();
                auto guard = task.guard(*this, coordinator);
                auto err = scan_error_t{info.get_path(), opt.assume_error()};
                task.push(new model::diff::local::io_failure_t(std::move(err)));
                return;
            } else {
                auto &chunk = opt.assume_value();
                if (chunk.data.size()) {
                    using request_t = hasher::payload::digest_request_t;
                    request_via<request_t>(hasher_proxy, reply_addr, std::move(chunk.data), (size_t)chunk.block_index,
                                           r::message_ptr_t(&message))
                        .send(init_timeout);
                    ++requested_hashes;
                }
            }
        }
    }
    return;
}

void scan_actor_t::on_hash(hasher::message::digest_response_t &res) noexcept {
    --requested_hashes;

    auto &rp = res.payload.req->payload.request_payload;
    auto msg = static_cast<message::rehash_needed_t *>(rp.custom.get());
    auto &info = msg->payload;
    info.ack_hashing();

    if (res.payload.ee) {
        auto &ee = res.payload.ee;
        auto file = info.get_file();
        LOG_ERROR(log, "on_hash, file: {}, block = {}, error: {}", file->get_full_name(), rp.block_index,
                  ee->message());
        return do_shutdown(ee);
    }

    bool queued_next = false;
    auto &digest = res.payload.res.digest;
    auto block_index = rp.block_index;
    info.ack_block(digest, block_index);
    hash_next(*msg, address);
    bool can_process_more = requested_hashes < requested_hashes_limit;
    auto &task = *info.get_task().get();
    auto guard = task.guard(*this, coordinator);
    queued_next = !can_process_more;
    if (info.is_complete()) {
        if (info.has_valid_blocks()) {
            task.push(new model::diff::local::blocks_availability_t(*info.get_file(), info.valid_blocks()));
        } else {
            auto &file = *info.get_file();
            LOG_DEBUG(log, "removing temporal of '{}' as it corrupted", file.get_full_name());
            auto r = info.remove();
            if (r) {
                auto err = scan_error_t{info.get_path(), r.assume_error()};
                task.push(new model::diff::local::io_failure_t(std::move(err)));
            }
        }
    }

    if (!queued_next && !requested_hashes) {
        send<payload::scan_progress_t>(address, info.get_task());
    }
}

void scan_actor_t::commit_new_file(new_chunk_iterator_t &info) noexcept {
    assert(info.is_complete());
    auto &hashes = info.get_hashes();
    auto folder_id = std::string(info.get_task()->get_folder_id());
    auto &file = info.get_metadata();
    proto::set_block_size(file, info.get_block_size());
    int offset = 0;

    // file.clear_blocks();
    for (auto &b : hashes) {
        auto block_info = proto::BlockInfo{offset, b.size, b.digest, b.weak};
        proto::add_blocks(file, std::move(block_info));
        offset += b.size;
    }

    auto &task = *info.get_task().get();
    auto guard = task.guard(*this, coordinator);
    task.push(new model::diff::advance::local_update_t(*cluster, *sequencer, std::move(file), std::move(folder_id)));
}

void scan_actor_t::on_hash_new(hasher::message::digest_response_t &res) noexcept {
    --requested_hashes;

    auto &rp = res.payload.req->payload.request_payload;
    auto msg = static_cast<message::hash_anew_t *>(rp.custom.get());
    auto &info = msg->payload;

    auto &path = info.get_path();
    if (res.payload.ee) {
        auto &ee = res.payload.ee;
        LOG_ERROR(log, "on_hash_new, file: {}, block = {}, error: {}", path.string(), rp.block_index, ee->message());
        return do_shutdown(ee);
    }

    auto &hash_info = res.payload.res;
    auto block_size = rp.data.size();

    info.ack(rp.block_index, hash_info.weak, hash_info.digest, block_size);

    while (info.has_more_chunks() && (requested_hashes < requested_hashes_limit)) {
        hash_next(*msg, new_files);
    }
    if (info.is_complete()) {
        commit_new_file(info);
        send<payload::scan_progress_t>(address, info.get_task());
    }
}

void scan_actor_t::post_scan(scan_task_t &task) noexcept {
    auto folder_id = task.get_folder_id();
    auto folder = cluster->get_folders().by_id(folder_id);
    auto self_fi = folder->get_folder_infos().by_device(*cluster->get_device());
    auto &self_files = self_fi->get_file_infos();
    auto aug = folder->get_augmentation();
    auto folder_entity = dynamic_cast<presentation::folder_entity_t *>(aug.get());
    auto &seen = task.get_seen_paths();
    if (folder_entity) {
        using queue_t = std::pmr::list<presentation::entity_t *>;
        using F = syncspirit::presentation::presence_t::features_t;
        auto buffer = std::array<std::byte, 10 * 1024>();
        auto pool = std::pmr::monotonic_buffer_resource(buffer.data(), buffer.size());
        auto allocator = std::pmr::polymorphic_allocator<char>(&pool);
        auto queue = queue_t(allocator);
        {
            auto &top_level_children = folder_entity->get_children();
            auto inserter = std::back_insert_iterator(queue);
            std::copy(top_level_children.begin(), top_level_children.end(), inserter);
        }
        while (!queue.empty()) {
            auto entity = queue.front();
            queue.pop_front();
            auto best = entity->get_best();
            if (best) {
                auto file_name = entity->get_path().get_full_name();
                auto it = seen.end();
                auto f = best->get_features();
                if (f & F::deleted) {
                    it = seen.find(file_name);
                    if ((it == seen.end()) && !self_files.by_name(file_name)) {
                        using local_update_t = model::diff::advance::local_update_t;
                        auto presence = static_cast<const presentation::cluster_file_presence_t *>(best);
                        auto &peer_file = presence->get_file_info();
                        auto pr_file = peer_file.as_proto(true);
                        LOG_DEBUG(log, "assuming deleted '{}' in the folder '{}'", file_name, folder_id);
                        task.push(new local_update_t(*cluster, *sequencer, std::move(pr_file), folder_id));
                    }
                }
                if (f & F::directory) {
                    if (it == seen.end()) {
                        it = seen.find(file_name);
                    }
                    auto recurse = [&]() -> bool {
                        bool creation_allowed = false;
                        if (it != seen.end()) {
                            auto &path = it->second;
                            auto ec = sys::error_code();
                            auto status = bfs::status(path, ec);
                            if (!ec) {
                                using perms_t = bfs::perms;
                                creation_allowed = (status.type() == bfs::file_type::directory) &&
                                                   ((status.permissions() & perms_t::owner_write) != perms_t::none);
                            }
                        } else {
                            creation_allowed = f & F::deleted;
                        }
                        return creation_allowed;
                    }();
                    if (recurse) {
                        auto &children = entity->get_children();
                        auto inserter = std::back_insert_iterator(queue);
                        std::copy(children.begin(), children.end(), inserter);
                    }
                }
            }
        }
    }
}


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

#pragma once

#include "model/cluster.h"
#include "model/messages.h"
#include "model/misc/sequencer.h"
#include "model/diff/cluster_visitor.h"
#include "utils/log.h"
#include "hasher/messages.h"
#include "messages.h"
#include "scan_task.h"
#include <rotor.hpp>

namespace syncspirit {
namespace fs {

namespace r = rotor;
namespace outcome = boost::outcome_v2;

struct SYNCSPIRIT_API scan_actor_config_t : r::actor_config_t {
    config::fs_config_t fs_config;
    model::cluster_ptr_t cluster;
    model::sequencer_ptr_t sequencer;
    uint32_t requested_hashes_limit;
};

template <typename Actor> struct scan_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 &&fs_config(const config::fs_config_t &value) && noexcept {
        parent_t::config.fs_config = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

    builder_t &&requested_hashes_limit(uint32_t value) && noexcept {
        parent_t::config.requested_hashes_limit = 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 &&sequencer(model::sequencer_ptr_t value) && noexcept {
        parent_t::config.sequencer = std::move(value);
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }
};

struct SYNCSPIRIT_API scan_actor_t : public r::actor_base_t, private model::diff::cluster_visitor_t {
    using config_t = scan_actor_config_t;
    template <typename Actor> using config_builder_t = scan_actor_config_builder_t<Actor>;

    explicit scan_actor_t(config_t &cfg);

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

  private:
    using clock_t = r::pt::microsec_clock;

    scan_errors_t initiate_hash(scan_task_ptr_t task, const bfs::path &path, proto::FileInfo &metadata) noexcept;
    void commit_new_file(new_chunk_iterator_t &info) noexcept;

    void on_model_update(model::message::model_update_t &message) noexcept;
    void on_scan(message::scan_progress_t &message) noexcept;
    void on_hash(hasher::message::digest_response_t &res) noexcept;
    void on_rehash(message::rehash_needed_t &message) noexcept;
    void on_hash_anew(message::hash_anew_t &message) noexcept;
    void on_hash_new(hasher::message::digest_response_t &res) noexcept;
    void post_scan(scan_task_t &task) noexcept;

    outcome::result<void> operator()(const model::diff::local::scan_start_t &, void *custom) noexcept override;

    template <typename Message> void hash_next(Message &m, const r::address_ptr_t &reply_addr) noexcept;

    model::cluster_ptr_t cluster;
    model::sequencer_ptr_t sequencer;
    config::fs_config_t fs_config;
    r::address_ptr_t hasher_proxy;
    r::address_ptr_t new_files; /* for routing */
    utils::logger_t log;
    r::address_ptr_t coordinator;
    uint32_t requested_hashes_limit;
    uint32_t requested_hashes = 0;
    uint32_t progress = 0;
};

} // namespace fs
} // namespace syncspirit

repo_name: basiliscos/syncspirit

license: mit


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

#include "scan_scheduler.h"
#include "net/names.h"
#include "model/diff/modify/upsert_folder.h"
#include "model/diff/local/scan_finish.h"
#include "model/diff/local/scan_request.h"
#include "model/diff/local/scan_start.h"
#include "model/diff/local/synchronization_finish.h"

using namespace syncspirit::fs;

scan_scheduler_t::scan_scheduler_t(config_t &cfg)
    : r::actor_base_t(cfg), cluster{std::move(cfg.cluster)}, scan_in_progress{false} {}

void scan_scheduler_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::names::fs_scheduler, false);
        log = utils::get_logger(identity);
    });
    plugin.with_casted<r::plugin::registry_plugin_t>([&](auto &p) {
        p.register_name(net::names::fs_scheduler, address);
        p.discover_name(net::names::coordinator, coordinator, true).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(&scan_scheduler_t::on_model_update, coordinator);
                plugin->subscribe_actor(&scan_scheduler_t::on_ui_ready, coordinator);
            }
        });
        p.discover_name(net::names::fs_scanner, fs_scanner, true).link();
    });
}

void scan_scheduler_t::on_ui_ready(model::message::ui_ready_t &) noexcept {
    LOG_TRACE(log, "on_ui_ready");
    scan_next();
}

void scan_scheduler_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, nullptr);
    if (!r) {
        auto ee = make_error(r.assume_error());
        do_shutdown(ee);
    }
}

auto scan_scheduler_t::operator()(const model::diff::modify::upsert_folder_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (!scan_in_progress) {
        scan_next_or_schedule();
    }
    return diff.visit_next(*this, custom);
}

auto scan_scheduler_t::operator()(const model::diff::local::scan_request_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    scan_queue.emplace_back(diff.folder_id);
    if (!scan_in_progress) {
        scan_next_or_schedule();
    }
    return diff.visit_next(*this, custom);
}

auto scan_scheduler_t::operator()(const model::diff::local::scan_finish_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    scan_in_progress = false;
    scan_next_or_schedule();
    return diff.visit_next(*this, custom);
}

auto scan_scheduler_t::operator()(const model::diff::local::synchronization_finish_t &diff, void *custom) noexcept
    -> outcome::result<void> {
    if (!scan_in_progress) {
        scan_next_or_schedule();
    }
    return diff.visit_next(*this, custom);
}

void scan_scheduler_t::scan_next_or_schedule() noexcept {
    auto next = scan_next();
    if (!next) {
        return;
    }
    bool do_start_timer = false;
    if (timer_id) {
        if (schedule_option->at > next.value().at) {
            LOG_TRACE(log, "cancelling previous schedule");
            cancel_timer(*timer_id);
            do_start_timer = true;
        }
    } else {
        do_start_timer = true;
    }
    if (do_start_timer) {
        LOG_DEBUG(log, "scheduling folders after {}s", next->interval.total_seconds());
        timer_id = start_timer(next->interval, *this, &scan_scheduler_t::on_timer);
        schedule_option = next;
    }
}

auto scan_scheduler_t::scan_next() noexcept -> schedule_option_t {
    while (!scan_queue.empty()) {
        auto folder_id = scan_queue.front();
        scan_queue.pop_front();
        auto folder = cluster->get_folders().by_id(folder_id);
        if (!folder || folder->is_synchronizing() || folder->is_suspended()) {
            continue;
        }
        for (auto it = scan_queue.begin(); it != scan_queue.end();) {
            if (*it == folder_id) {
                it = scan_queue.erase(it);
            } else {
                ++it;
            }
        }
        initiate_scan(folder_id);
        return {};
    }

    auto folder = model::folder_ptr_t();
    auto deadline = r::pt::ptime{};
    auto now = r::pt::second_clock::local_time();
    for (auto it : cluster->get_folders()) {
        auto &f = it.item;
        auto interval = f->get_rescan_interval();
        if (!interval || f->is_scanning() || f->is_synchronizing() || f->is_suspended()) {
            continue;
        }
        auto interval_s = r::pt::seconds{interval};
        auto prev_scan = f->get_scan_finish();
        auto it_deadline = prev_scan.is_not_a_date_time() ? now : prev_scan + interval_s;
        auto select_it = !folder || it_deadline < deadline ||
                         ((it_deadline == deadline) && (folder->get_rescan_interval() > interval));
        if (select_it) {
            folder = f;
            deadline = it_deadline;
        }
    }

    if (folder && deadline <= now) {
        initiate_scan(folder->get_id());
        return {};
    }
    if (folder) {
        auto folder_id = std::string(folder->get_id());
        return next_schedule_t{deadline - now, deadline, folder_id};
    }
    return {};
}

void scan_scheduler_t::on_timer(r::request_id_t, bool cancelled) noexcept {
    timer_id = {};
    if (!cancelled) {
        auto &folder_id = schedule_option->folder_id;
        if (auto folder = cluster->get_folders().by_id(folder_id); folder) {
            initiate_scan(folder->get_id());
        }
    }
}

void scan_scheduler_t::initiate_scan(std::string_view folder_id) noexcept {
    LOG_DEBUG(log, "initiating folder '{}' scan", folder_id);
    auto diff = model::diff::cluster_diff_ptr_t{};
    auto now = r::pt::microsec_clock::local_time();
    diff = new model::diff::local::scan_start_t(folder_id, now);
    send<model::payload::model_update_t>(coordinator, std::move(diff));
    scan_in_progress = true;
}


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

#pragma once

#include "model/cluster.h"
#include "model/messages.h"
#include "model/diff/cluster_visitor.h"
#include "utils/log.h"
#include <rotor.hpp>
#include <optional>
#include <list>

namespace syncspirit {
namespace fs {

namespace r = rotor;
namespace outcome = boost::outcome_v2;

struct SYNCSPIRIT_API scan_scheduler_config_t : r::actor_config_t {
    model::cluster_ptr_t cluster;
};

template <typename Actor> struct scan_scheduler_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));
    }
};

struct SYNCSPIRIT_API scan_scheduler_t : public r::actor_base_t, private model::diff::cluster_visitor_t {
    using config_t = scan_scheduler_config_t;
    template <typename Actor> using config_builder_t = scan_scheduler_config_builder_t<Actor>;

    explicit scan_scheduler_t(config_t &cfg);

    void configure(r::plugin::plugin_base_t &plugin) noexcept override;
    void on_ui_ready(model::message::ui_ready_t &) noexcept;

  private:
    struct next_schedule_t {
        r::pt::time_duration interval;
        r::pt::ptime at;
        std::string folder_id;
    };
    using schedule_option_t = std::optional<next_schedule_t>;
    using scan_queue_t = std::list<std::string>;

    void on_model_update(model::message::model_update_t &message) noexcept;
    void on_timer(r::request_id_t, bool cancelled) noexcept;

    schedule_option_t scan_next() noexcept;
    void scan_next_or_schedule() noexcept;
    void initiate_scan(std::string_view folder_id) noexcept;

    outcome::result<void> operator()(const model::diff::modify::upsert_folder_t &, void *custom) noexcept override;
    outcome::result<void> operator()(const model::diff::local::scan_request_t &, void *custom) noexcept override;
    outcome::result<void> operator()(const model::diff::local::scan_finish_t &, void *custom) noexcept override;
    outcome::result<void> operator()(const model::diff::local::synchronization_finish_t &,
                                     void *custom) noexcept override;

    model::cluster_ptr_t cluster;
    scan_queue_t scan_queue;
    utils::logger_t log;
    r::address_ptr_t fs_scanner;
    r::address_ptr_t coordinator;
    std::optional<r::request_id_t> timer_id;
    schedule_option_t schedule_option;
    bool scan_in_progress;
};

} // namespace fs
} // namespace syncspirit


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

#include "scan_task.h"
#include "utils.h"
#include "model/messages.h"
#include "fs/messages.h"
#include <boost/nowide/convert.hpp>

using namespace syncspirit::fs;

scan_task_t::scan_task_t(model::cluster_ptr_t cluster_, std::string_view folder_id_,
                         const config::fs_config_t &config_) noexcept
    : folder_id{folder_id_}, cluster{cluster_}, config{config_}, current_diff{nullptr} {
    auto &fm = cluster->get_folders();
    folder = fm.by_id(folder_id);
    if (!folder) {
        return;
    }

    auto my_folder = folder->get_folder_infos().by_device(*cluster->get_device());
    if (!my_folder) {
        return;
    }

    auto path = folder->get_path();
    dirs_queue.push_back(path);

    root = path;

    auto &orig_files = my_folder->get_file_infos();
    for (auto &it : orig_files) {
        files.put(it.item);
    }

    bytes_left = config.bytes_scan_iteration_limit;
    files_left = config.files_scan_iteration_limit;
    assert(bytes_left > 0);
    assert(files_left > 0);

    log = utils::get_logger("fs.scan_task");
}

scan_task_t::~scan_task_t() {}

const std::string &scan_task_t::get_folder_id() const noexcept { return folder_id; }

scan_result_t scan_task_t::advance() noexcept {
    if (!unknown_files_queue.empty()) {
        auto &file = unknown_files_queue.front();
        auto &&r = advance_unknown_file(file);
        unknown_files_queue.pop_front();
        return r;
    }
    if (!files_queue.empty()) {
        auto &file = files_queue.front();
        auto &&r = advance_file(file);
        files_queue.pop_front();
        return r;
    } else if (!dirs_queue.empty()) {
        auto &path = dirs_queue.front();
        auto &&r = advance_dir(path);
        dirs_queue.pop_front();
        return r;
    }
    if (!dirs_queue.empty()) {
        return true;
    }
    if (files.size() != 0) {
        auto file = files.begin()->item;
        seen_paths.insert({std::string(file->get_name()), file->get_path()});
        files.remove(file);
        if (file->is_deleted()) {
            return unchanged_meta_t{std::move(file)};
        } else {
            return removed_t{std::move(file)};
        }
    }

    return false;
}

scan_result_t scan_task_t::advance_dir(const bfs::path &dir) noexcept {
    sys::error_code ec;

    bool exists = bfs::exists(dir, ec);
    if (ec) {
        return scan_errors_t{scan_error_t{dir, ec}};
    }
    if (!exists) {
        return true;
    }

    auto record_path = [&](const bfs::path &file) {
        auto relative = relativize(file, root);
        auto str = boost::nowide::narrow(relative.generic_wstring());
        seen_paths.insert({std::move(str), file});
    };

    scan_errors_t errors;
    auto it = bfs::directory_iterator(dir, ec);
    auto removed = model::file_infos_map_t{};
    if (ec) {
        auto str = std::string{};
        if (dir != root) {
            str = bfs::relative(dir, root).string() + "/";
        }
        for (auto &it : files) {
            auto &file = *it.item;
            bool remove = str.empty() || (file.get_name().find(str) == 0);
            if (remove) {
                seen_paths.insert({std::string(file.get_name()), file.get_path()});
                removed.put(it.item);
            }
        }
        seen_paths.insert({str, dir});
        errors.push_back(scan_error_t{dir, ec});
    } else {
        for (; it != bfs::directory_iterator(); ++it) {
            auto &child = *it;
            record_path(child);
            sys::error_code ec;
            auto status = bfs::symlink_status(child, ec);
            if (ec && (status.type() != bfs::file_type::symlink)) {
                errors.push_back(scan_error_t{child, ec});
                continue;
            }
            auto rp = relativize(child, root).generic_string();
            auto file = files.by_name(rp);
            if (file) {
                files_queue.push_back(file);
                removed.put(file);
                if (status.type() == bfs::file_type::directory) {
                    dirs_queue.push_back(child);
                }
                continue;
            }

            auto file_type = status.type();
            proto::FileInfo metadata;
            proto::set_name(metadata, rp);

            if (file_type == bfs::file_type::regular || file_type == bfs::file_type::directory) {
                auto modification_time = bfs::last_write_time(child, ec);
                if (ec) {
                    errors.push_back(scan_error_t{dir, ec});
                    continue;
                }
                proto::set_modified_s(metadata, to_unix(modification_time));

                auto permissions = static_cast<uint32_t>(status.permissions());
                proto::set_permissions(metadata, permissions);
            }
            if (file_type == bfs::file_type::regular) {
                proto::set_type(metadata, proto::FileInfoType::FILE);
                auto sz = bfs::file_size(child, ec);
                if (ec) {
                    errors.push_back(scan_error_t{dir, ec});
                    continue;
                }
                proto::set_size(metadata, sz);
            } else if (file_type == bfs::file_type::directory) {
                proto::set_type(metadata, proto::FileInfoType::DIRECTORY);
                dirs_queue.push_back(child);
            } else if (file_type == bfs::file_type::symlink) {
                auto target = bfs::read_symlink(child, ec);
                if (ec) {
                    errors.push_back(scan_error_t{dir, ec});
                    continue;
                }
                proto::set_symlink_target(metadata, target.string());
                proto::set_type(metadata, proto::FileInfoType::SYMLINK);
            } else {
                LOG_WARN(log, "unknown/unimplemented file type {} : {}", (int)status.type(), bfs::path(child).string());
                continue;
            }

            unknown_files_queue.push_back(unknown_file_t{child, std::move(metadata)});
        }
    }
    for (auto &it : removed) {
        files.remove(it.item);
    }
    if (!errors.empty()) {
        return errors;
    }

    return true;
}

scan_result_t scan_task_t::advance_file(file_info_t &file) noexcept {
    if (file->is_file()) {
        return advance_regular_file(file);
    } else if (file->is_dir()) {
        return unchanged_meta_t{file};
    } else {
        assert(file->is_link());
        return advance_symlink_file(file);
    }
}

scan_result_t scan_task_t::advance_regular_file(file_info_t &file) noexcept {
    sys::error_code ec;

    auto path = file->get_path();
    auto meta = proto::FileInfo();
    bool changed = false;

    auto sz = bfs::file_size(path, ec);
    if (ec) {
        return file_error_t{path, ec};
    }

    proto::set_size(meta, sz);
    if (sz != (size_t)file->get_size()) {
        changed = true;
    }

    auto modified = bfs::last_write_time(path, ec);
    if (ec) {
        return file_error_t{path, ec};
    }
    auto modified_s = to_unix(modified);
    proto::set_modified_s(meta, modified_s);
    if (modified_s != file->get_modified_s()) {
        changed = true;
    }

    auto status = bfs::status(path, ec);
    if (ec) {
        return file_error_t{path, ec};
    }
    auto permissions = static_cast<uint32_t>(status.permissions());
    proto::set_permissions(meta, permissions);
    if (permissions != file->get_permissions()) {
        changed = true;
    }

    if (changed) {
        using FT = proto::FileInfoType;
        proto::set_name(meta, file->get_name());
        proto::set_type(meta, FT::FILE);
        return changed_meta_t{file, std::move(meta)};
    }

    return unchanged_meta_t{file};
}

scan_result_t scan_task_t::advance_symlink_file(file_info_t &file) noexcept {
    auto path = file->get_path();

    if (!bfs::is_symlink(path)) {
        LOG_CRITICAL(log, "not implemented change tracking: symlink -> non-symblink");
        return unchanged_meta_t{file};
    }

    sys::error_code ec;
    auto target = bfs::read_symlink(path, ec);
    if (ec) {
        return file_error_t{path, ec};
    }

    auto target_str = target.string();
    if (target_str == file->get_link_target()) {
        return unchanged_meta_t{file};
    } else {
        using FT = proto::FileInfoType;
        auto meta = proto::FileInfo();
        proto::set_name(meta, file->get_name());
        proto::set_type(meta, FT::SYMLINK);
        proto::set_symlink_target(meta, std::move(target_str));
        return changed_meta_t{file, std::move(meta)};
    }
}

scan_result_t scan_task_t::advance_unknown_file(unknown_file_t &file) noexcept {
    auto &path = file.path;
    if (!is_temporal(file.path.filename())) {
        return unknown_file_t{std::move(path), std::move(file.metadata)};
    }

    auto peer_file = model::file_info_ptr_t{};
    auto peer_counter = proto::Counter();
    auto relative_path = [&]() -> std::string {
        auto rp = relativize(path, root);
        auto name = path.filename();
        auto name_str = name.string();
        auto new_name = name_str.substr(0, name_str.size() - tmp_suffix.size());
        auto new_path = rp.parent_path() / new_name;
        return new_path.generic_string();
    }();
    for (auto &it : folder->get_folder_infos()) {
        auto &folder_info = it.item;
        if (folder_info->get_device() == cluster->get_device()) {
            continue;
        }
        auto &files = folder_info->get_file_infos();
        auto f = files.by_name(relative_path);
        if (f) {
            seen_paths.insert({std::string(f->get_name()), path});
            if (!peer_file) {
                peer_file = std::move(f);
                peer_counter = peer_file->get_version()->get_best();
            } else {
                auto &c = f->get_version()->get_best();
                if (proto::get_value(peer_counter) < proto::get_value(c)) {
                    peer_counter = c;
                    peer_file = std::move(f);
                    break;
                }
            }
        }
    }

    sys::error_code ec;
    if (!peer_file) {
        LOG_INFO(log, "source cannot be found for temporal file {}, removing orphan", relative_path);
        bfs::remove(path, ec);
        return orphaned_removed_t{path};
    }

    auto modified_time = bfs::last_write_time(path, ec);
    if (ec) {
        LOG_DEBUG(log, "removing outdated temporally {}, cannot get last modification: {}", path.string(),
                  ec.message());
        bfs::remove(path, ec);
        return file_error_t{path, ec};
    }
    auto modified_at = to_unix(modified_time);
    auto now = std::time(nullptr);
    if (modified_at + config.temporally_timeout <= now) {
        LOG_DEBUG(log, "removing outdated temporally {}", path.string());
        bfs::remove(path, ec);
        if (ec) {
            return file_error_t{path, ec};
        }
        return incomplete_removed_t{peer_file};
    }

    auto actual_size = proto::get_size(file.metadata);
    bool size_matches = static_cast<std::int64_t>(peer_file->get_size()) == actual_size;

    if (!size_matches) {
        LOG_DEBUG(log, "removing temporally '{}' because of size-mismatch or outdated source", path.string());
        bfs::remove(path, ec);
        if (ec) {
            return file_error_t{path, ec};
        }
        return incomplete_removed_t{peer_file};
    }

    auto opt = file_t::open_read(path);
    if (!opt) {
        LOG_DEBUG(log, "try to remove temporally {}, which cannot open ", path.string());
        bfs::remove(path, ec);
        if (ec) {
            return file_error_t{path, ec};
        }
        return incomplete_removed_t{peer_file};
    }

    auto &opened_file = opt.assume_value();
    return incomplete_t{peer_file, file_ptr_t(new file_t(std::move(opened_file)))};
}

void scan_task_t::push(model::diff::cluster_diff_t *update, std::int64_t bytes_consumed) noexcept {
    if (current_diff) {
        current_diff = current_diff->assign_sibling(update);
    } else {
        update_diff.reset(update);
        current_diff = update;
    }
    bytes_left -= bytes_consumed;
    --files_left;
}

auto scan_task_t::get_seen_paths() const noexcept -> const seen_paths_t & { return seen_paths; }

auto scan_task_t::guard(r::actor_base_t &actor, r::address_ptr_t coordinator) noexcept -> send_guard_t {
    return send_guard_t(*this, actor, coordinator);
}

scan_task_t::send_guard_t::send_guard_t(scan_task_t &task_, r::actor_base_t &actor_,
                                        r::address_ptr_t coordinator_) noexcept
    : task{task_}, actor{actor_}, coordinator{coordinator_}, force_send{false}, manage_progress{false} {}

void scan_task_t::send_guard_t::send_by_force() noexcept { force_send = true; }

void scan_task_t::send_guard_t::send_progress() noexcept { manage_progress = true; }

scan_task_t::send_guard_t::~send_guard_t() {
    auto consume = force_send || task.bytes_left <= 0 || task.files_left <= 0;
    if (consume) {
        auto diff = std::move(task.update_diff);
        if (diff) {
            task.current_diff = nullptr;
            task.bytes_left = task.config.bytes_scan_iteration_limit;
            task.files_left = task.config.files_scan_iteration_limit;
            actor.send<model::payload::model_update_t>(coordinator, std::move(diff), nullptr);
            if (manage_progress) {
                auto &sup = actor.get_supervisor();
                auto address = actor.get_address();
                auto message = rotor::make_routed_message<payload::scan_progress_t>(coordinator, address, &task);
                sup.put(message);
            }
        }
    } else if (manage_progress) {
        actor.send<payload::scan_progress_t>(actor.get_address(), &task);
    }
}

//


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

#pragma once

#include "syncspirit-export.h"
#include "model/cluster.h"
#include "model/diff/local/io_failure.h"
#include "config/fs.h"
#include "utils/log.h"
#include "utils/string_comparator.hpp"
#include "file.h"
#include "proto/proto-helpers.h"
#include <rotor.hpp>
#include <boost/intrusive_ptr.hpp>
#include <boost/smart_ptr/intrusive_ref_counter.hpp>
#include <filesystem>
#include <list>
#include <variant>
#include <cstdint>
#include <unordered_map>

namespace syncspirit::fs {

namespace r = rotor;
namespace bfs = std::filesystem;
namespace sys = boost::system;

using scan_error_t = model::diff::local::io_error_t;
using scan_errors_t = model::diff::local::io_errors_t;

struct unchanged_meta_t {
    model::file_info_ptr_t file;
};

struct changed_meta_t {
    model::file_info_ptr_t file;
    proto::FileInfo metadata;
};

struct incomplete_t {
    model::file_info_ptr_t file;
    file_ptr_t opened_file;
};

struct incomplete_removed_t {
    model::file_info_ptr_t file;
};

struct orphaned_removed_t {
    bfs::path path;
};

struct removed_t {
    model::file_info_ptr_t file;
};

struct file_error_t {
    bfs::path path;
    sys::error_code ec;
};

struct unknown_file_t {
    bfs::path path;
    proto::FileInfo metadata;
};

using scan_result_t = std::variant<bool, scan_errors_t, changed_meta_t, unchanged_meta_t, incomplete_t, removed_t,
                                   incomplete_removed_t, orphaned_removed_t, unknown_file_t, file_error_t>;

struct SYNCSPIRIT_API scan_task_t : boost::intrusive_ref_counter<scan_task_t, boost::thread_unsafe_counter> {
    using file_info_t = model::file_info_ptr_t;
    using path_queue_t = std::list<bfs::path>;
    using files_queue_t = std::list<file_info_t>;
    using unknown_files_queue_t = std::list<unknown_file_t>;
    using seen_paths_t = std::unordered_map<std::string, bfs::path, utils::string_hash_t, utils::string_eq_t>;

    struct send_guard_t {
        send_guard_t(scan_task_t &task, r::actor_base_t &actor, r::address_ptr_t coordinator) noexcept;
        ~send_guard_t();

        void send_by_force() noexcept;
        void send_progress() noexcept;

        scan_task_t &task;
        r::actor_base_t &actor;
        r::address_ptr_t coordinator;
        bool force_send;
        bool manage_progress;
    };

    scan_task_t(model::cluster_ptr_t cluster, std::string_view folder_id, const config::fs_config_t &config) noexcept;
    ~scan_task_t();

    void push(model::diff::cluster_diff_t *model_update, std::int64_t bytes_consumed = 0) noexcept;
    send_guard_t guard(r::actor_base_t &actor, r::address_ptr_t coordinator) noexcept;
    scan_result_t advance() noexcept;
    const std::string &get_folder_id() const noexcept;
    const seen_paths_t &get_seen_paths() const noexcept;

  private:
    scan_result_t advance_dir(const bfs::path &dir) noexcept;
    scan_result_t advance_file(file_info_t &file) noexcept;
    scan_result_t advance_regular_file(file_info_t &file) noexcept;
    scan_result_t advance_symlink_file(file_info_t &file) noexcept;
    scan_result_t advance_unknown_file(unknown_file_t &file) noexcept;

    std::string folder_id;
    model::folder_ptr_t folder;
    model::cluster_ptr_t cluster;
    model::file_infos_map_t files;
    utils::logger_t log;
    config::fs_config_t config;
    std::int64_t bytes_left;
    std::int64_t files_left;

    path_queue_t dirs_queue;
    files_queue_t files_queue;
    unknown_files_queue_t unknown_files_queue;
    bfs::path root;
    seen_paths_t seen_paths;

    model::diff::cluster_diff_ptr_t update_diff;
    model::diff::cluster_diff_t *current_diff;

    friend struct send_guard_t;
};

using scan_task_ptr_t = boost::intrusive_ptr<scan_task_t>;

} // namespace syncspirit::fs


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

#include "utils.h"
#include <cstdint>

namespace syncspirit::fs {

const std::string_view tmp_suffix = ".syncspirit-tmp";

static const std::int64_t _block_sizes[] = {
    // clang-format off
    (1 << 7) * 1024ull,
    (1 << 8) * 1024ull,
    (1 << 9) * 1024ull,
    (1 << 10) * 1024ull,
    (1 << 11) * 1024ull,
    (1 << 12) * 1024ull,
    (1 << 13) * 1024ull,
    (1 << 14) * 1024ull,
    // clang-format on
};

const std::size_t block_sizes_sz = 8;
const std::int64_t *block_sizes = _block_sizes;

static const constexpr size_t max_blocks_count = 2000;

block_division_t get_block_size(int64_t sz, int32_t prev_size) noexcept {
    auto bs = std::int64_t{0};
    if (block_sizes[0] <= sz) {
        for (size_t i = 0; i < block_sizes_sz; ++i) {
            if (block_sizes[i] == prev_size) {
                bs = prev_size;
                break;
            }
        }
    }

    if (!bs) {
        for (size_t i = 0; i < block_sizes_sz; ++i) {
            if (block_sizes[i] * static_cast<std::int64_t>(max_blocks_count) >= sz) {
                bs = block_sizes[i];
                if (bs > sz) {
                    bs = sz;
                }
                break;
            }
        }
    }
    if (bs == 0 && sz) {
        bs = block_sizes[block_sizes_sz - 1];
    }

    auto count = std::int64_t{0};
    if (bs != 0) {
        count = sz / bs;
        if (count * bs < sz) {
            ++count;
        }
    }

    return {static_cast<size_t>(count), (int32_t)bs};
}

bfs::path make_temporal(const bfs::path &path) noexcept {
    auto copy = path;
    copy += tmp_suffix.data();
    return copy;
}

bool is_temporal(const bfs::path &path) noexcept {
    auto name = path.string();
    if (name.length() < tmp_suffix.length()) {
        return false;
    }
    auto pos = name.find(tmp_suffix, name.length() - tmp_suffix.length());
    return (pos != name.npos);
}

bfs::path relativize(const bfs::path &path, const bfs::path &root) noexcept {
    auto it_path = path.begin();
    auto it_root = root.begin();

    while (it_path != path.end() && it_root != root.end() && *it_path == *it_root) {
        ++it_path;
        ++it_root;
    }

    auto sub = bfs::path();
    while (it_path != path.end()) {
        sub /= *it_path;
        ++it_path;
    }

    return sub;
}

using seconds_t = std::chrono::seconds;
using sys_clock_t = std::chrono::system_clock;

std::int64_t to_unix(const fs_time_t &at) {
    auto sys_at = fs_time_t::clock::to_sys(at);
    return std::chrono::duration_cast<seconds_t>(sys_at.time_since_epoch()).count();
}

fs_time_t from_unix(std::int64_t at) {
    auto sys_time = sys_clock_t::from_time_t(at);
    return fs_time_t::clock::from_sys(sys_time);
}

} // namespace syncspirit::fs


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

#pragma once

#include <boost/outcome.hpp>
#include <filesystem>
#include <boost/smart_ptr/local_shared_ptr.hpp>
#include "syncspirit-export.h"

namespace syncspirit {
namespace fs {

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

using fs_time_t = std::filesystem::file_time_type;

struct block_division_t {
    size_t count;
    int32_t size;
};

SYNCSPIRIT_API bfs::path make_temporal(const bfs::path &path) noexcept;
SYNCSPIRIT_API bool is_temporal(const bfs::path &path) noexcept;
SYNCSPIRIT_API block_division_t get_block_size(int64_t file_size, int32_t prev_size) noexcept;
SYNCSPIRIT_API bfs::path relativize(const bfs::path &path, const bfs::path &root) noexcept;
SYNCSPIRIT_API std::int64_t to_unix(const fs_time_t &at);
SYNCSPIRIT_API fs_time_t from_unix(std::int64_t at);

SYNCSPIRIT_API extern const std::size_t block_sizes_sz;
SYNCSPIRIT_API extern const std::int64_t *block_sizes;
SYNCSPIRIT_API extern const std::string_view tmp_suffix;

} // namespace fs
} // namespace syncspirit


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

#include "hasher_actor.h"
#include "../utils/tls.h"
#include <fmt/core.h>
#include <zlib.h>
#include <algorithm>

using namespace syncspirit::hasher;
static const constexpr size_t SZ = SHA256_DIGEST_LENGTH;

hasher_actor_t::hasher_actor_t(config_t &cfg) : r::actor_base_t(cfg), index(cfg.index) {}

void hasher_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(fmt::format("hasher-{}", index), false);
        log = utils::get_logger(identity);
    });
    plugin.with_casted<r::plugin::registry_plugin_t>([&](auto &p) { p.register_name(identity, get_address()); });

    plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
        p.subscribe_actor(&hasher_actor_t::on_digest);
        p.subscribe_actor(&hasher_actor_t::on_validation);
    });
}

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

void hasher_actor_t::shutdown_finish() noexcept {
    LOG_TRACE(log, "{}, shutdown_finish", identity);
    get_supervisor().shutdown();
    r::actor_base_t::shutdown_finish();
}

void hasher_actor_t::on_digest(message::digest_request_t &req) noexcept {
    LOG_TRACE(log, "{}, on_digest", identity);

    unsigned char digest[SZ];
    auto &data = req.payload.request_payload.data;

    utils::digest(data.data(), data.size(), digest);

    // alder32
    auto weak_hash = adler32(0L, Z_NULL, 0);
    weak_hash = adler32(weak_hash, (const unsigned char *)data.data(), data.size());

    reply_to(req, utils::bytes_t(digest, digest + SZ), static_cast<uint32_t>(weak_hash));
}

void hasher_actor_t::on_validation(message::validation_request_t &req) noexcept {
    LOG_TRACE(log, "{}, on_validation", identity);
    auto &payload = *req.payload.request_payload;
    auto &h = payload.hash;

    unsigned char digest[SZ];
    auto &data = payload.data;

    utils::digest(data.data(), data.size(), digest);
    bool eq = std::equal(h.begin(), h.end(), digest, digest + SZ);

    reply_to(req, eq);
}


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

#pragma once

#include "utils/log.h"
#include "messages.h"
#include "syncspirit-export.h"

#include <rotor.hpp>

namespace syncspirit {
namespace hasher {

struct hasher_actor_config_t : r::actor_config_t {
    uint32_t index;
};

template <typename Actor> struct hasher_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 &&index(uint32_t value) && noexcept {
        parent_t::config.index = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }
};

struct SYNCSPIRIT_API hasher_actor_t : public r::actor_base_t {
    using config_t = hasher_actor_config_t;
    template <typename Actor> using config_builder_t = hasher_actor_config_builder_t<Actor>;

    explicit hasher_actor_t(config_t &cfg);

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

  private:
    void on_digest(message::digest_request_t &req) noexcept;
    void on_validation(message::validation_request_t &req) noexcept;

    utils::logger_t log;
    uint32_t index;
};

} // namespace hasher
} // namespace syncspirit


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

#include "hasher_proxy_actor.h"
#include "../utils/error_code.h"
#include <fmt/core.h>
#include <numeric>

using namespace syncspirit::hasher;

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

hasher_proxy_actor_t::hasher_proxy_actor_t(config_t &config) : r::actor_base_t(config) {
    hashers.resize(config.hasher_threads);
    hasher_scores.resize(config.hasher_threads);
    hasher_threads = config.hasher_threads;
    name = config.name;
}

void hasher_proxy_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(name, false);
        log = utils::get_logger(identity);
    });
    plugin.with_casted<r::plugin::registry_plugin_t>([&](auto &p) {
        p.register_name(name, get_address());
        for (size_t i = 0; i < hashers.size(); ++i) {
            auto name = fmt::format("hasher-{}", i + 1);
            p.discover_name(name, hashers[i], true).link(false);
        }
    });
    plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
        p.subscribe_actor(&hasher_proxy_actor_t::on_validation_request);
        p.subscribe_actor(&hasher_proxy_actor_t::on_validation_response);
        p.subscribe_actor(&hasher_proxy_actor_t::on_digest_request);
        p.subscribe_actor(&hasher_proxy_actor_t::on_digest_response);
    });
}

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

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

r::address_ptr_t hasher_proxy_actor_t::find_next_hasher() noexcept {
    uint32_t score = std::numeric_limits<uint32_t>::max();
    uint32_t min = 0;
    for (uint32_t j = 0; j < hasher_threads; ++j) {
        uint32_t i = (index + j) % hasher_threads;
        if (hasher_scores[i] == 0) {
            min = i;
            break;
        }
        if (hasher_scores[i] < score) {
            score = hasher_scores[i];
            min = i;
        }
    }
    ++hasher_scores[min];
    index = (index + 1) % hasher_threads;
    resources->acquire(resource::hash);
    return hashers[min];
}

void hasher_proxy_actor_t::free_hasher(r::address_ptr_t &addr) noexcept {
    resources->release(resource::hash);
    for (uint32_t i = 0; i < hashers.size(); ++i) {
        if (hashers[i] == addr) {
            --hasher_scores[i];
            return;
        }
    }
    assert(0 && "should not happen");
}

void hasher_proxy_actor_t::on_validation_request(hasher::message::validation_request_t &req) noexcept {
    LOG_TRACE(log, "{}, on_validation_request", identity);
    auto hasher = find_next_hasher();
    auto &p = *req.payload.request_payload;
    request<hasher::payload::validation_request_t>(hasher, p.data, p.hash, &req).send(init_timeout);
}

void hasher_proxy_actor_t::on_validation_response(hasher::message::validation_response_t &res) noexcept {
    using request_t = hasher::message::validation_request_t;
    LOG_TRACE(log, "{}, on_validation_response", identity);
    auto req = (request_t *)res.payload.req->payload.request_payload->custom.get();
    auto &payload = res.payload;
    auto &ee = payload.ee;
    if (ee) {
        reply_with_error(*req, std::move(ee));
    } else {
        reply_to(*req, payload.res.valid);
    }
    free_hasher(payload.req->address);
}

void hasher_proxy_actor_t::on_digest_request(hasher::message::digest_request_t &req) noexcept {
    LOG_TRACE(log, "{}, on_digest_request", identity);
    auto hasher = find_next_hasher();
    auto &p = req.payload.request_payload;
    request<hasher::payload::digest_request_t>(hasher, p.data, p.block_index, &req).send(init_timeout);
}

void hasher_proxy_actor_t::on_digest_response(hasher::message::digest_response_t &res) noexcept {
    LOG_TRACE(log, "{}, on_digest_response", identity);
    using request_t = hasher::message::digest_request_t;
    auto req = (request_t *)res.payload.req->payload.request_payload.custom.get();
    auto &payload = res.payload;
    auto &ee = payload.ee;
    if (ee) {
        reply_with_error(*req, std::move(ee));
    } else {
        reply_to(*req, payload.res.digest, payload.res.weak);
    }
    free_hasher(payload.req->address);
}


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

#pragma once

#include "messages.h"
#include "utils/log.h"
#include "syncspirit-export.h"

namespace syncspirit {
namespace hasher {

namespace r = rotor;

struct hasher_proxy_actor_config_t : r::actor_config_t {
    uint32_t hasher_threads;
    std::string name;
};

template <typename Actor> struct hasher_proxy_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 &&hasher_threads(uint32_t value) && noexcept {
        parent_t::config.hasher_threads = value;
        return std::move(*static_cast<typename parent_t::builder_t *>(this));
    }

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

struct SYNCSPIRIT_API hasher_proxy_actor_t : public r::actor_base_t {
    using config_t = hasher_proxy_actor_config_t;
    template <typename Actor> using config_builder_t = hasher_proxy_actor_config_builder_t<Actor>;

    hasher_proxy_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 addresses_t = std::vector<r::address_ptr_t>;

    void on_digest_request(hasher::message::digest_request_t &req) noexcept;
    void on_digest_response(hasher::message::digest_response_t &res) noexcept;
    void on_validation_request(hasher::message::validation_request_t &req) noexcept;
    void on_validation_response(hasher::message::validation_response_t &res) noexcept;

    r::address_ptr_t find_next_hasher() noexcept;
    void free_hasher(r::address_ptr_t &addr) noexcept;

    utils::logger_t log;
    addresses_t hashers;
    std::vector<uint32_t> hasher_scores;
    uint32_t hasher_threads;
    std::string name;
    uint32_t index = 0;
};

} // namespace hasher
} // namespace syncspirit


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

#include "hasher_supervisor.h"
#include "hasher_actor.h"
#include "../net/names.h"

using namespace syncspirit::hasher;

hasher_supervisor_t::hasher_supervisor_t(config_t &config) : parent_t(config), index{config.index} {}

void hasher_supervisor_t::configure(r::plugin::plugin_base_t &plugin) noexcept {
    parent_t::configure(plugin);
    plugin.with_casted<r::plugin::address_maker_plugin_t>([&](auto &p) {
        p.set_identity("hasher.supervisor", false);
        log = utils::get_logger(identity);
    });
    plugin.with_casted<r::plugin::registry_plugin_t>(
        [&](auto &p) { p.discover_name(net::names::coordinator, coordinator, true).link(false); });
    plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &) { launch(); });
}

void hasher_supervisor_t::launch() noexcept {
    create_actor<hasher_actor_t>().index(index).timeout(shutdown_timeout).finish();
}

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


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

#pragma once

#include "config/main.h"
#include "utils/log.h"
#include "syncspirit-export.h"

#include <rotor/thread.hpp>

namespace syncspirit {
namespace hasher {

namespace r = rotor;
namespace rth = rotor::thread;

struct hasher_supervisor_config_t : r::supervisor_config_t {
    uint32_t index;
};

template <typename Supervisor> struct hasher_supervisor_config_builder_t : r::supervisor_config_builder_t<Supervisor> {
    using builder_t = typename Supervisor::template config_builder_t<Supervisor>;
    using parent_t = r::supervisor_config_builder_t<Supervisor>;
    using parent_t::parent_t;

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

struct SYNCSPIRIT_API hasher_supervisor_t : rth::supervisor_thread_t {
    using parent_t = rth::supervisor_thread_t;
    using config_t = hasher_supervisor_config_t;
    template <typename Supervisor> using config_builder_t = hasher_supervisor_config_builder_t<Supervisor>;

    hasher_supervisor_t(config_t &config);

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

  private:
    void launch() noexcept;

    uint32_t index;
    utils::logger_t log;
    r::address_ptr_t coordinator;
};

} // namespace hasher
} // namespace syncspirit


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

#pragma once

#include "utils/bytes.h"
#include <rotor.hpp>

namespace syncspirit {
namespace hasher {

namespace r = rotor;

namespace payload {

struct digest_response_t {
    utils::bytes_t digest;
    uint32_t weak;
};

struct digest_request_t {
    using response_t = digest_response_t;
    utils::bytes_t data;
    size_t block_index;
    r::message_ptr_t custom;
};

struct validation_response_t {
    bool valid;
};

struct validation_request_t : r::arc_base_t<validation_request_t> {
    using response_t = validation_response_t;
    utils::bytes_view_t data;
    utils::bytes_t hash;
    r::message_ptr_t custom;

    validation_request_t(utils::bytes_view_t data_, utils::bytes_t hash_, r::message_ptr_t custom_ = nullptr) noexcept
        : data{data_}, hash{std::move(hash_)}, custom{std::move(custom_)} {}
};

} // namespace payload

namespace message {

using digest_request_t = r::request_traits_t<payload::digest_request_t>::request::message_t;
using digest_response_t = r::request_traits_t<payload::digest_request_t>::response::message_t;

using validation_request_t = r::request_traits_t<payload::validation_request_t>::request::message_t;
using validation_response_t = r::request_traits_t<payload::validation_request_t>::response::message_t;

} // namespace message

} // namespace hasher
} // namespace syncspirit


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

#include "block_info.h"
#include "file_info.h"
#include "proto/proto-helpers.h"
#include "db/prefix.h"
#include "misc/error_code.h"
#include <spdlog/spdlog.h>

namespace syncspirit::model {

static const constexpr char prefix = (char)(db::prefix::block_info);

auto block_info_t::make_strict_hash(utils::bytes_view_t hash) noexcept -> strict_hash_t {
    assert(hash.size() <= digest_length);
    auto r = strict_hash_t{};
    r.data[0] = prefix;
    memset(r.data + 1, 0, digest_length);
    memcpy(r.data + 1, hash.data(), hash.size());
    return r;
}

auto block_info_t::strict_hash_t::get_hash() noexcept -> utils::bytes_view_t {
    return utils::bytes_view_t(data + 1, digest_length);
}

auto block_info_t::strict_hash_t::get_key() noexcept -> utils::bytes_view_t {
    return utils::bytes_view_t(data, data_length);
}

block_info_t::block_info_t(utils::bytes_view_t key) noexcept { std::copy(key.begin(), key.end(), hash); }

block_info_t::block_info_t(const proto::BlockInfo &block) noexcept
    : weak_hash{proto::get_weak_hash(block)}, size{proto::get_size(block)} {
    hash[0] = prefix;
}

template <> void block_info_t::assign<db::BlockInfo>(const db::BlockInfo &block) noexcept {
    weak_hash = db::get_weak_hash(block);
    size = db::get_size(block);
}

outcome::result<block_info_ptr_t> block_info_t::create(utils::bytes_view_t key, const db::BlockInfo &data) noexcept {
    if (key.size() != data_length) {
        return make_error_code(error_code_t::invalid_block_key_length);
    }
    if (key[0] != prefix) {
        return make_error_code(error_code_t::invalid_block_prefix);
    }

    auto ptr = block_info_ptr_t(new block_info_t(key));
    ptr->assign(data);
    return outcome::success(ptr);
}

outcome::result<block_info_ptr_t> block_info_t::create(const proto::BlockInfo &block) noexcept {
    auto h = proto::get_hash(block);
    if (h.size() > digest_length) {
        return make_error_code(error_code_t::invalid_block_key_length);
    }

    auto ptr = block_info_ptr_t(new block_info_t(block));
    auto &h_ptr = ptr->hash;
    std::copy(h.begin(), h.end(), h_ptr + 1);
    auto left = digest_length - h.size();
    if (left) {
        std::fill_n(h_ptr + 1 + h.size(), left, 0);
    }
    return outcome::success(ptr);
}

proto::BlockInfo block_info_t::as_bep(size_t offset) const noexcept {
    proto::BlockInfo r;
    proto::set_size(r, size);
    proto::set_weak_hash(r, weak_hash);
    proto::set_hash(r, get_hash());
    proto::set_offset(r, offset);
    return r;
}

utils::bytes_t block_info_t::serialize() const noexcept { return db::encode(db::BlockInfo{weak_hash, size}); }

void block_info_t::link(file_info_t *file_info, size_t block_index) noexcept {
    file_blocks.emplace_back(this, file_info, block_index);
}

auto block_info_t::unlink(file_info_t *file_info) noexcept -> removed_incides_t {
    removed_incides_t r;
    for (auto it = file_blocks.begin(); it != file_blocks.end();) {
        auto &fb = *it;
        if (fb.matches(this, file_info)) {
            r.push_back(fb.block_index());
            it = file_blocks.erase(it);
        } else {
            ++it;
        }
    }
    assert(!r.empty() && "at least one block has been removed");
    return r;
}

void block_info_t::mark_local_available(file_info_t *file_info) noexcept {
    auto predicate = [&](file_block_t &block) { return block.matches(this, file_info); };
    auto it = std::find_if(file_blocks.begin(), file_blocks.end(), predicate);
    assert(it != file_blocks.end());
    it->mark_locally_available();
}

file_block_t block_info_t::local_file() noexcept {
    for (auto &b : file_blocks) {
        if (b.is_locally_available()) {
            return b;
        }
    }
    return {};
}

bool block_info_t::is_locked() const noexcept { return locked != 0; }

void block_info_t::lock() noexcept {
    assert(!locked);
    ++locked;
}

void block_info_t::unlock() noexcept {
    assert(locked);
    --locked;
}

template <> SYNCSPIRIT_API utils::bytes_view_t get_index<0>(const block_info_ptr_t &item) noexcept {
    return item->get_hash();
}

block_info_ptr_t block_infos_map_t::by_hash(utils::bytes_view_t hash) const noexcept { return get(std::move(hash)); }

} // namespace syncspirit::model


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

#pragma once

#include "misc/arc.hpp"
#include "misc/map.hpp"
#include "misc/file_block.h"
#include "proto/proto-fwd.hpp"
#include "utils/bytes.h"
#include "syncspirit-export.h"
#include <boost/outcome.hpp>
#include <cstdint>

namespace syncspirit::model {

namespace outcome = boost::outcome_v2;

struct file_info_t;
using file_info_ptr_t = intrusive_ptr_t<file_info_t>;
struct block_info_t;
using block_info_ptr_t = intrusive_ptr_t<block_info_t>;

struct SYNCSPIRIT_API block_info_t final : arc_base_t<block_info_t> {
    using removed_incides_t = std::vector<size_t>;
    using file_blocks_t = std::vector<file_block_t>;
    static const constexpr size_t digest_length = 32;
    static const constexpr size_t data_length = digest_length + 1;

    struct strict_hash_t {
        unsigned char data[data_length];
        utils::bytes_view_t get_hash() noexcept;
        utils::bytes_view_t get_key() noexcept;
    };

    static strict_hash_t make_strict_hash(utils::bytes_view_t hash) noexcept;

    static outcome::result<block_info_ptr_t> create(utils::bytes_view_t key, const db::BlockInfo &data) noexcept;
    static outcome::result<block_info_ptr_t> create(const proto::BlockInfo &block) noexcept;

    inline utils::bytes_view_t get_hash() const noexcept { return utils::bytes_view_t(hash + 1, digest_length); }
    inline utils::bytes_view_t get_key() const noexcept { return utils::bytes_view_t(hash); }
    inline std::uint32_t get_weak_hash() const noexcept { return weak_hash; }
    inline std::uint32_t get_size() const noexcept { return size; }
    inline size_t usages() const noexcept { return file_blocks.size(); }
    inline file_blocks_t &get_file_blocks() { return file_blocks; }
    inline const file_blocks_t &get_file_blocks() const { return file_blocks; }

    proto::BlockInfo as_bep(size_t offset) const noexcept;
    utils::bytes_t serialize() const noexcept;

    void link(file_info_t *file_info, size_t block_index) noexcept;
    removed_incides_t unlink(file_info_t *file_info) noexcept;

    void mark_local_available(file_info_t *file_info) noexcept;
    file_block_t local_file() noexcept;

    bool is_locked() const noexcept;
    void lock() noexcept;
    void unlock() noexcept;

    inline bool operator==(const block_info_t &right) const noexcept {
        auto lh = get_hash();
        auto rh = right.get_hash();
        return std::equal(lh.begin(), lh.end(), rh.begin(), rh.end());
    }
    inline bool operator!=(const block_info_t &right) const noexcept { return !(*this == right); }

  private:
    template <typename T> void assign(const T &item) noexcept;
    block_info_t(utils::bytes_view_t key) noexcept;
    block_info_t(const proto::BlockInfo &block) noexcept;

    unsigned char hash[data_length];
    std::uint32_t weak_hash = 0;
    std::int32_t size = 0;
    file_blocks_t file_blocks;
    std::uint32_t locked = 0;
};

struct SYNCSPIRIT_API block_infos_map_t : generic_map_t<block_info_ptr_t, 1> {
    block_info_ptr_t by_hash(utils::bytes_view_t view) const noexcept;
};

} // namespace syncspirit::model


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

#include "cluster.h"
#include "proto/proto-helpers.h"
#include <spdlog/spdlog.h>

using namespace syncspirit;
using namespace syncspirit::model;

cluster_t::cluster_t(device_ptr_t device_, int32_t write_requests_) noexcept
    : device(device_), tainted{false}, write_requests{write_requests_} {}

proto::ClusterConfig cluster_t::generate(const device_t &target) const noexcept {
    proto::ClusterConfig r;
    for (auto it : folders) {
        auto &folder = it.item;
        auto folder_opt = folder->generate(target);
        if (folder_opt) {
            auto &folder_value = folder_opt.value();
            proto::add_folders(r, std::move(folder_value));
        }
    }
    return r;
}

devices_map_t &cluster_t::get_devices() noexcept { return devices; }

const devices_map_t &cluster_t::get_devices() const noexcept { return devices; }

ignored_devices_map_t &cluster_t::get_ignored_devices() noexcept { return ignored_devices; }

const ignored_devices_map_t &cluster_t::get_ignored_devices() const noexcept { return ignored_devices; }

ignored_folders_map_t &cluster_t::get_ignored_folders() noexcept { return ignored_folders; }

block_infos_map_t &cluster_t::get_blocks() noexcept { return blocks; }

const block_infos_map_t &cluster_t::get_blocks() const noexcept { return blocks; }

folders_map_t &cluster_t::get_folders() noexcept { return folders; }

auto cluster_t::get_pending_devices() noexcept -> pending_devices_map_t & { return pending_devices; }

auto cluster_t::get_pending_devices() const noexcept -> const pending_devices_map_t & { return pending_devices; }

auto cluster_t::get_pending_folders() noexcept -> pending_folder_map_t & { return pending_folders; }

auto cluster_t::get_pending_folders() const noexcept -> const pending_folder_map_t & { return pending_folders; }

const folders_map_t &cluster_t::get_folders() const noexcept { return folders; }

int32_t cluster_t::get_write_requests() const noexcept { return write_requests; }

void cluster_t::modify_write_requests(int32_t delta) noexcept {
    write_requests += delta;
    assert(write_requests >= 0);
}


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

#pragma once

#include "misc/arc.hpp"
#include "device.h"
#include "ignored_device.h"
#include "ignored_folder.h"
#include "folder.h"
#include "pending_device.h"
#include "pending_folder.h"
#include "block_info.h"

namespace syncspirit::model {

struct SYNCSPIRIT_API cluster_t final : arc_base_t<cluster_t> {

    cluster_t(device_ptr_t device_, int32_t write_requests) noexcept;
    cluster_t(const cluster_t &) = delete;

    proto::ClusterConfig generate(const model::device_t &target) const noexcept;
    inline const device_ptr_t &get_device() const noexcept { return device; }
    block_infos_map_t &get_blocks() noexcept;
    const block_infos_map_t &get_blocks() const noexcept;
    devices_map_t &get_devices() noexcept;
    const devices_map_t &get_devices() const noexcept;
    const ignored_devices_map_t &get_ignored_devices() const noexcept;
    ignored_devices_map_t &get_ignored_devices() noexcept;
    ignored_folders_map_t &get_ignored_folders() noexcept;
    folders_map_t &get_folders() noexcept;
    pending_devices_map_t &get_pending_devices() noexcept;
    const pending_devices_map_t &get_pending_devices() const noexcept;
    pending_folder_map_t &get_pending_folders() noexcept;

    const folders_map_t &get_folders() const noexcept;
    const pending_folder_map_t &get_pending_folders() const noexcept;
    inline bool is_tainted() const noexcept { return tainted; }
    inline void mark_tainted() noexcept { tainted = true; }
    int32_t get_write_requests() const noexcept;
    void modify_write_requests(int32_t delta) noexcept;

  private:
    device_ptr_t device;
    folders_map_t folders;
    block_infos_map_t blocks;
    devices_map_t devices;
    ignored_devices_map_t ignored_devices;
    ignored_folders_map_t ignored_folders;
    pending_folder_map_t pending_folders;
    pending_devices_map_t pending_devices;
    bool tainted = false;
    int32_t write_requests;
};

using cluster_ptr_t = intrusive_ptr_t<cluster_t>;

} // namespace syncspirit::model


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

#include "device.h"
#include "proto/proto-helpers-db.h"
#include "db/prefix.h"
#include "misc/error_code.h"
#include "misc/file_iterator.h"
#include "utils/time.h"
#include "utils/error_code.h"
#include <cassert>

namespace syncspirit::model {

static const constexpr char prefix = (char)(db::prefix::device);

template <> auto device_t::assign(const db::Device &d) noexcept -> outcome::result<void> {
    name = db::get_name(d);
    compression = db::get_compression(d);
    cert_name = db::get_cert_name(d);
    introducer = db::get_introducer(d);
    auto_accept = db::get_auto_accept(d);
    paused = db::get_paused(d);
    skip_introduction_removals = db::get_skip_introduction_removals(d);
    static_uris.clear();
    uris.clear();
    last_seen = pt::from_time_t(db::get_last_seen(d));

    auto uris = uris_t{};
    auto addresses_count = db::get_addresses_size(d);
    for (size_t i = 0; i < addresses_count; ++i) {
        auto uri = utils::parse(db::get_addresses(d, i));
        if (!uri) {
            return make_error_code(utils::error_code_t::malformed_url);
        }
        assert(uri);
        uris.emplace_back(uri);
    }
    set_static_uris(std::move(uris));
    return outcome::success();
}

outcome::result<device_ptr_t> device_t::create(utils::bytes_view_t key, const db::Device &data) noexcept {
    if (key[0] != prefix) {
        return make_error_code(error_code_t::invalid_device_prefix);
    }

    auto id = device_id_t::from_sha256(key.subspan(1));
    if (!id) {
        return make_error_code(error_code_t::invalid_device_sha256_digest);
    }

    auto name = db::get_name(data);
    auto cert_name = db::get_cert_name(data);
    auto ptr = device_ptr_t(new device_t(id.value(), name, cert_name));
    auto ec = ptr->assign(data);
    if (!ec) {
        return ec.error();
    }
    return outcome::success(std::move(ptr));
}

outcome::result<device_ptr_t> device_t::create(const device_id_t &device_id, std::string_view name,
                                               std::string_view cert_name) noexcept {
    auto ptr = device_ptr_t(new device_t(device_id, name, cert_name));
    return outcome::success(std::move(ptr));
}

device_t::device_t(const device_id_t &device_id_, std::string_view name_, std::string_view cert_name_) noexcept
    : id(std::move(device_id_)), name{name_}, compression{proto::Compression::METADATA}, cert_name{cert_name_},
      introducer{false}, auto_accept{false}, paused{false}, skip_introduction_removals{false},
      state{device_state_t::offline}, last_seen{pt::from_time_t(0)}, rx_bytes{0}, tx_bytes{0} {}

device_t::~device_t() {}

auto device_t::update(const db::Device &source) noexcept -> outcome::result<void> { return assign(source); }

auto device_t::serialize(db::Device &r) const noexcept -> utils::bytes_t {
    db::set_name(r, name);
    db::set_compression(r, compression);
    if (cert_name.has_value()) {
        db::set_cert_name(r, cert_name.value());
    }
    db::set_introducer(r, introducer);
    db::set_skip_introduction_removals(r, skip_introduction_removals);
    db::set_auto_accept(r, auto_accept);
    db::set_paused(r, paused);
    db::set_last_seen(r, utils::as_seconds(last_seen));

    for (size_t i = 0; i < static_uris.size(); ++i) {
        auto buff = static_uris[i]->buffer();
        db::set_addresses(r, i, buff);
    }
    return db::encode(r);
}

auto device_t::serialize() const noexcept -> utils::bytes_t {
    db::Device r;
    return serialize(r);
}

void device_t::update_state(device_state_t new_state, std::string_view connection_id_) noexcept {
    if (state == device_state_t::online || new_state == device_state_t::online) {
        last_seen = pt::microsec_clock::local_time();
        if (new_state == device_state_t::online) {
            assert(!connection_id_.empty());
            bool new_wins = (connection_id.empty()) || (connection_id_.size() < connection_id.size()) ||
                            (connection_id_ < connection_id);
            if (new_wins) {
                connection_id = connection_id_;
            }
        }
    } else {
        connection_id = {};
    }
    state = new_state;
}

void device_t::update_contact(const tcp::endpoint &endpoint_, std::string_view client_name_,
                              std::string_view client_version_) noexcept {
    endpoint = endpoint_;
    client_name = client_name_;
    client_version = client_version_;
}

auto device_t::get_key() const noexcept -> utils::bytes_view_t { return id.get_key(); }

void device_t::set_static_uris(uris_t uris) noexcept { static_uris = std::move(uris); }

void device_t::assign_uris(const uris_t &uris_) noexcept { uris = uris_; }

auto device_t::create_iterator(cluster_t &cluster) noexcept -> file_iterator_ptr_t {
    assert(!iterator);
    iterator = new file_iterator_t(cluster, this);
    return iterator;
}

void device_t::release_iterator(file_iterator_ptr_t &it) noexcept {
    assert(it == iterator);
    (void)it;
    iterator.reset();
}

file_iterator_t *device_t::get_iterator() noexcept { return iterator.get(); }

std::string_view device_t::get_connection_id() noexcept { return connection_id; }

local_device_t::local_device_t(const device_id_t &device_id, std::string_view name, std::string_view cert_name) noexcept
    : device_t(device_id, name, cert_name) {
    state = device_state_t::online;
}

utils::bytes_view_t local_device_t::get_key() const noexcept { return local_device_id.get_key(); }

template <> SYNCSPIRIT_API utils::bytes_view_t get_index<0>(const device_ptr_t &item) noexcept {
    return item->get_key();
}

template <> SYNCSPIRIT_API utils::bytes_view_t get_index<1>(const device_ptr_t &item) noexcept {
    return item->device_id().get_sha256();
}

device_ptr_t devices_map_t::by_sha256(utils::bytes_view_t device_id) const noexcept { return get<1>(device_id); }

device_ptr_t devices_map_t::by_key(utils::bytes_view_t key) const noexcept { return get<0>(key); }

} // namespace syncspirit::model


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

#pragma once

#include "misc/augmentation.h"
#include "misc/map.hpp"
#include "device_id.h"
#include "remote_folder_info.h"
#include "utils/uri.h"
#include "utils/bytes.h"
#include "syncspirit-export.h"
#include "proto/proto-fwd.hpp"
#include <boost/asio.hpp>
#include <boost/outcome.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

namespace syncspirit::model {

namespace outcome = boost::outcome_v2;
namespace pt = boost::posix_time;

struct device_t;
struct file_iterator_t;
struct cluster_t;
using device_ptr_t = intrusive_ptr_t<device_t>;
using file_iterator_ptr_t = intrusive_ptr_t<file_iterator_t>;

enum class device_state_t { offline, discovering, connecting, online };

struct SYNCSPIRIT_API device_t : augmentable_t {
    using uris_t = utils::uri_container_t;
    using name_option_t = std::optional<std::string>;
    using tcp = boost::asio::ip::tcp;

    static outcome::result<device_ptr_t> create(utils::bytes_view_t key, const db::Device &data) noexcept;
    static outcome::result<device_ptr_t> create(const device_id_t &device_id, std::string_view name,
                                                std::string_view cert_name = "") noexcept;
    virtual ~device_t();

    virtual utils::bytes_view_t get_key() const noexcept;
    bool operator==(const device_t &other) const noexcept { return other.id == id; }
    bool operator!=(const device_t &other) const noexcept { return other.id != id; }

    utils::bytes_t serialize(db::Device &device) const noexcept;
    utils::bytes_t serialize() const noexcept;
    inline bool is_dynamic() const noexcept { return static_uris.empty(); }
    inline device_state_t get_state() const noexcept { return state; }
    void update_state(device_state_t new_state, std::string_view connection_id) noexcept;
    void update_contact(const tcp::endpoint &endpoint, std::string_view client_name,
                        std::string_view client_version) noexcept;
    inline device_id_t &device_id() noexcept { return id; }
    inline const device_id_t &device_id() const noexcept { return id; }
    inline std::string_view get_name() const noexcept { return name; }
    inline const name_option_t get_cert_name() const noexcept { return cert_name; }
    inline const tcp::endpoint &get_endpoint() const noexcept { return endpoint; }
    inline std::string_view get_client_name() const noexcept { return client_name; }
    inline std::string_view get_client_version() const noexcept { return client_version; }
    inline proto::Compression get_compression() const noexcept { return compression; }
    inline bool is_introducer() const noexcept { return introducer; }
    inline bool has_auto_accept() const noexcept { return auto_accept; }
    inline bool is_paused() const noexcept { return paused; }
    inline bool get_skip_introduction_removals() const noexcept { return skip_introduction_removals; }
    inline auto &get_remote_folder_infos() noexcept { return remote_folder_infos; }
    inline const pt::ptime &get_last_seen() const noexcept { return last_seen; }

    inline const uris_t &get_uris() const noexcept { return uris; }
    inline const uris_t &get_static_uris() const noexcept { return static_uris; }
    void set_static_uris(uris_t) noexcept;
    void assign_uris(const uris_t &uris) noexcept;

    std::string_view get_connection_id() noexcept;
    outcome::result<void> update(const db::Device &source) noexcept;

    inline size_t get_rx_bytes() const noexcept { return rx_bytes; }
    inline void set_rx_bytes(size_t value) noexcept { rx_bytes = value; }
    inline size_t get_tx_bytes() const noexcept { return tx_bytes; }
    inline void set_tx_bytes(size_t value) noexcept { tx_bytes = value; }

    file_iterator_ptr_t create_iterator(cluster_t &) noexcept;
    void release_iterator(file_iterator_ptr_t &) noexcept;
    file_iterator_t *get_iterator() noexcept;

  protected:
    device_t(const device_id_t &device_id, std::string_view name, std::string_view cert_name) noexcept;
    template <typename T> outcome::result<void> assign(const T &item) noexcept;

    device_id_t id;
    std::string name;
    proto::Compression compression;
    name_option_t cert_name;
    uris_t uris;
    uris_t static_uris;
    bool introducer;
    bool auto_accept;
    bool paused;
    bool skip_introduction_removals;
    device_state_t state = device_state_t::offline;
    tcp::endpoint endpoint;
    std::string client_name;
    std::string client_version;
    std::string connection_id;

    file_iterator_ptr_t iterator;

    remote_folder_infos_map_t remote_folder_infos;
    pt::ptime last_seen;
    std::size_t rx_bytes;
    std::size_t tx_bytes;
};

struct SYNCSPIRIT_API local_device_t final : device_t {
  public:
    local_device_t(const device_id_t &device_id, std::string_view name, std::string_view cert_name) noexcept;
    utils::bytes_view_t get_key() const noexcept override;
};

struct SYNCSPIRIT_API devices_map_t : public generic_map_t<device_ptr_t, 2> {
    device_ptr_t by_sha256(utils::bytes_view_t device_id) const noexcept;
    device_ptr_t by_key(utils::bytes_view_t key) const noexcept;
};

} // namespace syncspirit::model

namespace std {

template <> struct hash<syncspirit::model::device_t> {
    inline size_t operator()(const syncspirit::model::device_t &device) const noexcept {
        return std::hash<syncspirit::model::device_id_t>()(device.device_id());
    }
};

template <> struct hash<syncspirit::model::device_ptr_t> {
    inline size_t operator()(const syncspirit::model::device_ptr_t &device) const noexcept {
        return std::hash<syncspirit::model::device_t>()(*device);
    }
};

} // namespace std


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

#include "device_id.h"
#include <cassert>
#include "utils/base32.h"
#include "proto/luhn32.h"
#include "db/prefix.h"

#include <openssl/sha.h>
#include <boost/endian/conversion.hpp>

namespace be = boost::endian;

using namespace syncspirit::proto;
using namespace syncspirit::model;
using namespace syncspirit::utils;

static const constexpr char prefix = (char)(syncspirit::db::prefix::device);

device_id_t::device_id_t(std::string_view value_, bytes_view_t sha256_) noexcept : value{value_} {
    assert(sha256_.size() == digest_length);
    hash[0] = prefix;
    std::copy(sha256_.begin(), sha256_.end(), hash + 1);
}

device_id_t::device_id_t(device_id_t &&other) noexcept { *this = std::move(other); }

device_id_t::device_id_t(const device_id_t &other) noexcept {
    value = other.value;
    std::copy(other.hash, other.hash + data_length, hash);
}

device_id_t &device_id_t::operator=(device_id_t &&other) noexcept {
    value = std::move(other.value);
    std::copy(other.hash, other.hash + data_length, hash);
    return *this;
}

device_id_t &device_id_t::operator=(const device_id_t &other) noexcept {
    value = other.value;
    std::copy(other.hash, other.hash + data_length, hash);
    return *this;
}

std::string_view device_id_t::get_short() const noexcept { return std::string_view(value.data(), DASH_INT); }

bool device_id_t::matches(uint64_t value) const noexcept { return get_uint() == value; }

std::uint64_t device_id_t::get_uint() const noexcept {
    uint64_t id;
    auto ptr = hash + 1;
    std::copy(ptr, ptr + sizeof(id), reinterpret_cast<char *>(&id));
    return be::native_to_big(id);
}

std::optional<device_id_t> device_id_t::from_string(std::string_view value) noexcept {
    using result_t = std::optional<device_id_t>;
    char buff[DASHED_SIZE];
    if (value.size() != DASHED_SIZE) {
        return result_t{};
    }
    auto buff_end = buff + DASHED_SIZE;
    std::copy(value.begin(), value.end(), buff);

    // remove dashes
    for (auto i = buff, o = buff; i != buff_end;) {
        if (*i != '-') {
            *o++ = *i++;
        } else {
            i++;
        }
    }
    buff_end = buff + LUHNIZED_SIZE;
    *buff_end = 0; /* mostly for debugging purposes */
    for (auto i = buff; i < buff_end; i += CHECK_DIGIT_INT + 1) {
        std::string_view piece(i, CHECK_DIGIT_INT + 1);
        if (!proto::luhn32::validate(piece)) {
            return result_t{};
        }
    }

    // remove luhn check digits
    auto ptr = buff;
    for (size_t i = 0, c = 0; i < LUHNIZED_SIZE; ++i, ++c) {
        if (c == CHECK_DIGIT_INT) {
            c = -1;
            // SKIP
        } else {
            *ptr++ = buff[i];
        }
    }
    buff_end = buff + LUHNIZED_SIZE - LUHN_ITERATIONS;
    *buff_end = 0; /* mostly for debugging purposes */

    auto sha = base32::decode(std::string_view(buff, LUHNIZED_SIZE - LUHN_ITERATIONS));
    if (sha) {
        return device_id_t{value, sha.value()};
    }
    return result_t{};
}

std::optional<device_id_t> device_id_t::from_sha256(bytes_view_t sha256) noexcept {
    using result_t = std::optional<device_id_t>;
    auto sha256_enc = base32::encode(sha256);
    if (sha256_enc.size() != SHA256_B32_SIZE) {
        return result_t{};
    }

    auto sha_ptr = sha256_enc.data();
    char luhnized[LUHNIZED_SIZE];
    char *luhn_ptr = luhnized;
    for (std::size_t i = 0; i < LUHN_ITERATIONS; ++i) {
        std::memcpy(luhn_ptr, sha_ptr, CHECK_DIGIT_INT);
        auto piece_ptr = luhn_ptr;
        luhn_ptr += CHECK_DIGIT_INT;
        *luhn_ptr = luhn32::calculate(std::string_view(piece_ptr, CHECK_DIGIT_INT));
        ++luhn_ptr;
        sha_ptr += CHECK_DIGIT_INT;
    }

    char dashed[DASHED_SIZE];
    luhn_ptr = luhnized;
    char *dashed_ptr = dashed;
    for (std::size_t i = 0; i < DASH_ITERATIONS; ++i) {
        std::memcpy(dashed_ptr, luhn_ptr, DASH_INT);
        luhn_ptr += DASH_INT;
        dashed_ptr += DASH_INT;
        if (i < DASH_ITERATIONS - 1) {
            *dashed_ptr++ = '-';
        }
    }
    auto str = std::string_view(dashed, DASHED_SIZE);
    return device_id_t{str, sha256};
}

std::string device_id_t::make_short(std::uint64_t value) noexcept {
    if (value == 0) {
        return {};
    }
    auto be = be::native_to_big(value);
    auto view = utils::bytes_view_t(reinterpret_cast<const unsigned char *>(&be), sizeof(value));
    auto encoded = base32::encode(view.subspan(0, 5)); // it is enought to encode just first 5 bytes
    return encoded.substr(0, DASH_INT);
}

std::string device_id_t::make_short(utils::bytes_view_t sha_256) noexcept {
    std::uint64_t value;
    auto ptr = sha_256.data();
    std::copy(sha_256.data(), ptr + sizeof(value), reinterpret_cast<char *>(&value));
    be::native_to_big_inplace(value);

    auto view = utils::bytes_view_t(reinterpret_cast<const unsigned char *>(&value), sizeof(value));
    auto encoded = base32::encode(view.subspan(0, 5)); // it is enought to encode just first 5 bytes
    return encoded.substr(0, DASH_INT);
}

std::optional<device_id_t> device_id_t::from_cert(cert_data_view_t data) noexcept {
    using result_t = std::optional<device_id_t>;
    auto sha_result = sha256_digest(data);
    if (!sha_result) {
        return result_t{};
    }
    return from_sha256(sha_result.value());
}

namespace syncspirit::model {

static const unsigned char local_device_sha265[SHA256_DIGEST_LENGTH] = {0xFF};
static const cert_data_view_t local_device_cert(local_device_sha265, SHA256_DIGEST_LENGTH);
const device_id_t local_device_id = device_id_t::from_cert(local_device_cert).value();

} // namespace syncspirit::model


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

#pragma once

#include "utils/tls.h"
#include "utils/bytes.h"
#include "syncspirit-export.h"
#include <string>
#include <cstdint>
#include <optional>

namespace syncspirit::model {

struct SYNCSPIRIT_API device_id_t {
    static const constexpr std::size_t SHA256_B32_SIZE = 52;
    static const constexpr std::size_t CHECK_DIGIT_INT = 13;
    static const constexpr std::size_t LUHN_ITERATIONS = SHA256_B32_SIZE / 13;
    static const constexpr std::size_t LUHNIZED_SIZE = SHA256_B32_SIZE + LUHN_ITERATIONS;
    static const constexpr std::size_t DASH_INT = 7;
    static const constexpr std::size_t DASH_ITERATIONS = LUHNIZED_SIZE / DASH_INT;
    static const constexpr std::size_t DASHED_SIZE = LUHNIZED_SIZE + DASH_ITERATIONS - 1;

    static const constexpr size_t digest_length = 32;
    static const constexpr size_t data_length = digest_length + 1;

    static std::optional<device_id_t> from_string(std::string_view value) noexcept;
    static std::optional<device_id_t> from_sha256(utils::bytes_view_t sha_256) noexcept;
    static std::optional<device_id_t> from_cert(utils::cert_data_view_t cert) noexcept;
    static std::optional<device_id_t> from_uuid(std::string_view value) noexcept;
    static std::string make_short(std::uint64_t value) noexcept;
    static std::string make_short(utils::bytes_view_t sha_256) noexcept;

    device_id_t() noexcept {};
    device_id_t(device_id_t &&other) noexcept;
    device_id_t(const device_id_t &other) noexcept;

    device_id_t &operator=(device_id_t &&other) noexcept;
    device_id_t &operator=(const device_id_t &other) noexcept;

    bool operator==(const device_id_t &other) const noexcept { return get_sha256() == other.get_sha256(); }
    bool operator!=(const device_id_t &other) const noexcept { return !(*this == other); }
    explicit operator bool() const noexcept { return !value.empty(); }

    const std::string &get_value() const noexcept { return value; }

    std::string_view get_short() const noexcept;
    utils::bytes_view_t get_sha256() const noexcept { return utils::bytes_view_t(hash + 1, digest_length); }
    utils::bytes_view_t get_key() const noexcept { return utils::bytes_view_t(hash, data_length); }
    std::uint64_t get_uint() const noexcept;

    bool matches(uint64_t) const noexcept;

  private:
    device_id_t(const utils::cert_data_t &) noexcept;
    device_id_t(std::string_view value_, utils::bytes_view_t sha256_) noexcept;

    std::string value;
    unsigned char hash[data_length];
};

SYNCSPIRIT_API extern const device_id_t local_device_id;

} // namespace syncspirit::model

namespace std {

template <> struct hash<syncspirit::model::device_id_t> {
    inline size_t operator()(const syncspirit::model::device_id_t &device_id) const noexcept {
        return std::hash<std::string>()(device_id.get_value());
    }
};

template <> struct less<syncspirit::model::device_id_t> {
    using device_id_t = syncspirit::model::device_id_t;
    inline bool operator()(const device_id_t &lhs, const device_id_t &rhs) const noexcept {
        return lhs.get_value() < rhs.get_value();
    }
};

} // namespace std