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

#include <catch2/catch_all.hpp>
#include "test-utils.h"
#include "diff-builder.h"
#include "model/diff/peer/cluster_update.h"
#include "model/diff/contact/ignored_connected.h"
#include "model/diff/contact/unknown_connected.h"
#include "test_supervisor.h"
#include "access.h"
#include "model/cluster.h"
#include "db/utils.h"
#include "net/db_actor.h"
#include "access.h"
#include <filesystem>

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

namespace fs = std::filesystem;

namespace {
struct env {};
} // namespace

namespace syncspirit::net {

template <> inline auto &db_actor_t::access<env>() noexcept { return env; }

} // namespace syncspirit::net

namespace {

struct fixture_t {
    using load_msg_t = net::message::load_cluster_response_t;
    using load_msg_ptr_t = r::intrusive_ptr_t<load_msg_t>;
    using stats_msg_t = net::message::db_info_response_t;
    using stats_msg_ptr_t = r::intrusive_ptr_t<stats_msg_t>;

    fixture_t() noexcept : root_path{unique_path()}, path_quard{root_path} {
        test::init_logging();
        bfs::create_directory(root_path);
    }

    fixture_t(fixture_t &&source) noexcept
        : root_path(std::move(source.root_path)), path_quard(std::move(source.path_quard)) {}

    virtual supervisor_t::configure_callback_t configure() noexcept {
        return [&](r::plugin::plugin_base_t &plugin) {
            plugin.template with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
                p.subscribe_actor(r::lambda<load_msg_t>([&](load_msg_t &msg) {
                    auto &ee = msg.payload.ee;
                    if (ee) {
                        spdlog::error("cannot load cluster: {}", ee->message());
                        std::abort();
                    }
                    reply = &msg;
                    sup->send<model::payload::model_update_t>(db_addr, msg.payload.res.diff, nullptr);
                }));
                p.subscribe_actor(r::lambda<stats_msg_t>([&](stats_msg_t &msg) { stats_reply = &msg; }));
            });
        };
    }

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

        auto cluster = cluster_ptr_t(new cluster_t(my_device, 1));
        cluster->get_devices().put(my_device);

        auto peer_id = "VUV42CZ-IQD5A37-RPEBPM4-VVQK6E4-6WSKC7B-PVJQHHD-4PZD44V-ENC6WAZ";
        auto peer_device_id = device_id_t::from_string(peer_id).value();
        peer_device = device_t::create(peer_device_id, "peer-device").value();

        if (add_peer) {
            cluster->get_devices().put(peer_device);
        }
        return cluster;
    }

    virtual fixture_t &run() noexcept {
        cluster = make_cluster();

        r::system_context_t ctx;
        sup = ctx.create_supervisor<supervisor_t>().timeout(timeout).create_registry().finish();
        sup->cluster = cluster;
        sup->configure_callback = configure();

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

        auto db_config = config::db_config_t{1024 * 1024, 0, 64, 32};
        db_actor = sup->create_actor<db_actor_t>()
                       .cluster(cluster)
                       .db_dir(root_path.string())
                       .db_config(db_config)
                       .timeout(timeout)
                       .finish();
        sup->do_process();

        CHECK(static_cast<r::actor_base_t *>(db_actor.get())->access<to::state>() == r::state_t::OPERATIONAL);
        db_addr = db_actor->get_address();
        main();
        reply.reset();

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

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

    virtual void main() noexcept {}

    r::address_ptr_t db_addr;
    r::pt::time_duration timeout = r::pt::millisec{10};
    cluster_ptr_t cluster;
    device_ptr_t peer_device;
    device_ptr_t my_device;
    r::intrusive_ptr_t<supervisor_t> sup;
    r::intrusive_ptr_t<net::db_actor_t> db_actor;
    bfs::path root_path;
    path_guard_t path_quard;
    r::system_context_t ctx;
    load_msg_ptr_t reply;
    stats_msg_ptr_t stats_reply;
};

} // namespace

void test_db_population() {
    struct F : fixture_t {
        void main() noexcept override {
            auto &db_env = db_actor->access<env>();
            auto txn_opt = db::make_transaction(db::transaction_type_t::RW, db_env);
            REQUIRE(txn_opt);
            auto &txn = txn_opt.value();
            auto load_opt = db::load(db::prefix::device, txn);
            REQUIRE(load_opt);
            auto &values = load_opt.value();
            REQUIRE(values.size() == 1);
        }
    };
    F().run();
}

void test_loading_empty_db() {
    struct F : fixture_t {

        void main() noexcept override {
            sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
            sup->do_process();
            REQUIRE(reply);

            auto diff = reply->payload.res.diff;
            REQUIRE(diff->apply(*cluster, get_apply_controller()));

            auto devices = cluster->get_devices();
            REQUIRE(devices.size() == 2);
            REQUIRE(devices.by_sha256(my_device->device_id().get_sha256()));
            REQUIRE(devices.by_sha256(peer_device->device_id().get_sha256()));

            sup->request<net::payload::db_info_request_t>(sup->get_address()).send(timeout);
            sup->do_process();
            REQUIRE(stats_reply);
            auto &stats = stats_reply->payload.res;
            CHECK(stats.entries > 0);
            CHECK(stats.page_size > 0);
        }
    };

    F().run();
}

void test_folder_upserting() {
    struct F : fixture_t {
        void main() noexcept override {
            auto folder_id = "1234-5678";

            auto builder = diff_builder_t(*cluster);
            builder.upsert_folder(folder_id, "/my/path", "my-label").apply(*sup);

            auto folder = cluster->get_folders().by_id(folder_id);
            REQUIRE(folder);
            REQUIRE(folder->get_folder_infos().by_device(*cluster->get_device()));

            {
                sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                sup->do_process();
                REQUIRE(reply);
                auto cluster_clone = make_cluster();
                REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));

                auto folder_clone = cluster_clone->get_folders().by_id(folder->get_id());
                REQUIRE(folder_clone);
                REQUIRE(folder.get() != folder_clone.get());
                REQUIRE(folder_clone->get_label() == "my-label");
                REQUIRE(folder_clone->get_path().string() == "/my/path");
                REQUIRE(folder_clone->get_folder_infos().size() == 1);
                REQUIRE(folder_clone->get_folder_infos().by_device(*cluster->get_device()));
            }

            SECTION("upserting") {
                builder.upsert_folder(folder_id, "/my/path", "my-label-2").apply(*sup);
                REQUIRE(folder->get_label() == "my-label-2");

                sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                sup->do_process();
                REQUIRE(reply);

                auto cluster_clone = make_cluster();
                REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                auto folder_clone = cluster_clone->get_folders().by_id(folder->get_id());
                REQUIRE(folder_clone->get_label() == "my-label-2");
            }
        }
    };

    F().run();
}

void test_unknown_and_ignored_devices_1() {
    struct F : fixture_t {
        void main() noexcept override {
            auto d_id1 =
                device_id_t::from_string("LYXKCHX-VI3NYZR-ALCJBHF-WMZYSPK-QG6QJA3-MPFYMSO-U56GTUK-NA2MIAW").value();
            auto d_id2 =
                device_id_t::from_string("XBOWTOU-Y7H6RM6-D7WT3UB-7P2DZ5G-R6GNZG6-T5CCG54-SGVF3U5-LBM7RQB").value();

            db::SomeDevice sd_1;
            db::set_name(sd_1, "x1");
            auto unknown_device = pending_device_t::create(d_id1, sd_1).value();

            db::SomeDevice sd_2;
            db::set_name(sd_2, "x2");
            auto ignored_device = ignored_device_t::create(d_id2, sd_2).value();

            auto builder = diff_builder_t(*cluster);
            builder.add_unknown_device(d_id1, sd_1).add_ignored_device(d_id2, sd_2).apply(*sup);

            REQUIRE(cluster->get_pending_devices().size() == 1);
            REQUIRE(cluster->get_ignored_devices().size() == 1);

            {
                sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                sup->do_process();
                REQUIRE(reply);
                auto cluster_clone = make_cluster();
                REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                CHECK(cluster_clone->get_pending_devices().by_sha256(d_id1.get_sha256()));
                CHECK(cluster_clone->get_ignored_devices().by_sha256(d_id2.get_sha256()));
            }

            db::set_name(sd_1, "x1_2");
            db::set_name(sd_2, "x2_2");
            auto diff = model::diff::cluster_diff_ptr_t{};
            diff = new model::diff::contact::unknown_connected_t(*cluster, d_id1, sd_1);
            sup->send<model::payload::model_update_t>(sup->get_address(), std::move(diff), nullptr);
            diff = new model::diff::contact::ignored_connected_t(*cluster, d_id2, sd_2);
            sup->send<model::payload::model_update_t>(sup->get_address(), std::move(diff), nullptr);
            sup->do_process();

            {
                sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                sup->do_process();
                REQUIRE(reply);
                auto cluster_clone = make_cluster();
                REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                REQUIRE(cluster_clone->get_pending_devices().size() == 1);
                REQUIRE(cluster_clone->get_ignored_devices().size() == 1);
                auto unknown = cluster_clone->get_pending_devices().by_sha256(d_id1.get_sha256());
                auto ignored = cluster_clone->get_ignored_devices().by_sha256(d_id2.get_sha256());
                REQUIRE(unknown);
                REQUIRE(ignored);
                CHECK(unknown->get_name() == "x1_2");
                CHECK(ignored->get_name() == "x2_2");
            }

            builder.remove_unknown_device(*unknown_device).remove_ignored_device(*ignored_device).apply(*sup);
            REQUIRE(cluster->get_pending_devices().size() == 0);
            REQUIRE(cluster->get_ignored_devices().size() == 0);

            {
                sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                sup->do_process();
                REQUIRE(reply);
                auto cluster_clone = make_cluster();
                REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                REQUIRE(cluster_clone->get_pending_devices().size() == 0);
                REQUIRE(cluster_clone->get_ignored_devices().size() == 0);
            }
        }
    };
    F().run();
}

void test_unknown_and_ignored_devices_2() {
    struct F : fixture_t {
        void main() noexcept override {
            auto d_id =
                device_id_t::from_string("LYXKCHX-VI3NYZR-ALCJBHF-WMZYSPK-QG6QJA3-MPFYMSO-U56GTUK-NA2MIAW").value();

            db::SomeDevice sd;
            db::set_name(sd, "x1");
            auto builder = diff_builder_t(*cluster);

            builder.add_unknown_device(d_id, sd).apply(*sup);
            {
                sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                sup->do_process();
                REQUIRE(reply);
                auto cluster_clone = make_cluster();
                REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                CHECK(cluster_clone->get_pending_devices().by_sha256(d_id.get_sha256()));
                CHECK(!cluster_clone->get_ignored_devices().by_sha256(d_id.get_sha256()));
            }

            builder.add_ignored_device(d_id, sd).apply(*sup);
            {
                sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                sup->do_process();
                REQUIRE(reply);
                auto cluster_clone = make_cluster();
                REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                CHECK(!cluster_clone->get_pending_devices().by_sha256(d_id.get_sha256()));
                CHECK(cluster_clone->get_ignored_devices().by_sha256(d_id.get_sha256()));
            }
        }
    };
    F().run();
}

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

            auto builder = diff_builder_t(*cluster);
            builder.update_peer(peer_device->device_id(), "some_name", "some-cn", true).apply(*sup);

            auto sha256 = peer_device->device_id().get_sha256();
            auto device = cluster->get_devices().by_sha256(sha256);
            REQUIRE(device);
            CHECK(device->get_name() == "some_name");
            CHECK(device->get_cert_name() == "some-cn");

            sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
            sup->do_process();
            REQUIRE(reply);
            auto cluster_clone = make_cluster();
            REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));

            REQUIRE(cluster_clone->get_devices().size() == 2);
            auto device_clone = cluster_clone->get_devices().by_sha256(sha256);
            REQUIRE(device_clone);
            REQUIRE(device.get() != device_clone.get());
            CHECK(device_clone->get_name() == "some_name");
            CHECK(device_clone->get_cert_name() == "some-cn");
        }
    };

    F().run();
}

void test_folder_sharing() {
    struct F : fixture_t {
        void main() noexcept override {
            auto sha256 = peer_device->device_id().get_sha256();
            auto folder_id = "1234-5678";
            auto builder = diff_builder_t(*cluster);
            builder.update_peer(peer_device->device_id())
                .apply(*sup)
                .upsert_folder(folder_id, "/my/path")
                .apply(*sup)
                .configure_cluster(sha256)
                .add(sha256, folder_id, 5, 4)
                .finish()
                .apply(*sup);
            REQUIRE(cluster->get_pending_folders().size() == 1);

            builder.share_folder(sha256, folder_id).apply(*sup);
            REQUIRE(cluster->get_pending_folders().size() == 0);

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

            sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
            sup->do_process();
            REQUIRE(reply);
            auto cluster_clone = make_cluster();
            REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));

            auto peer_device = cluster_clone->get_devices().by_sha256(sha256);
            REQUIRE(peer_device);
            auto folder = cluster_clone->get_folders().by_id(folder_id);
            REQUIRE(folder);
            REQUIRE(folder->get_folder_infos().size() == 2);
            auto fi = folder->get_folder_infos().by_device(*peer_device);
            REQUIRE(fi);
            CHECK(fi->get_index() == 5);
            CHECK(fi->get_max_sequence() == 0);
            REQUIRE(cluster_clone->get_pending_folders().size() == 0);
        }
    };

    F().run();
}

void test_cluster_update_and_remove() {
    struct F : fixture_t {
        void main() noexcept override {
            auto sha256 = peer_device->device_id().get_sha256();
            auto folder_id = "1234-5678";
            auto unknown_folder_id = "5678-999";

            auto file = proto::FileInfo();
            proto::set_name(file, "a.txt");
            proto::set_size(file, 5ul);
            proto::set_block_size(file, 5ul);
            proto::set_sequence(file, 6ul);

            auto b_hash = utils::sha256_digest(as_bytes("12345")).value();
            auto &b = proto::add_blocks(file);
            proto::set_size(b, 5ul);
            proto::set_hash(b, b_hash);

            auto builder = diff_builder_t(*cluster);
            builder.update_peer(peer_device->device_id())
                .apply(*sup)
                .upsert_folder(folder_id, "/my/path")
                .configure_cluster(sha256)
                .add(sha256, folder_id, 5, proto::get_sequence(file))
                .add(sha256, unknown_folder_id, 5, 5)
                .finish()
                .apply(*sup)
                .share_folder(sha256, folder_id)
                .apply(*sup)
                .make_index(sha256, folder_id)
                .add(file, peer_device)
                .finish()
                .apply(*sup);

            REQUIRE(cluster->get_blocks().size() == 1);
            auto block = cluster->get_blocks().by_hash(b_hash);
            REQUIRE(block);

            auto folder = cluster->get_folders().by_id(folder_id);
            auto peer_folder_info = folder->get_folder_infos().by_device(*peer_device);

            REQUIRE(peer_folder_info);
            CHECK(peer_folder_info->get_max_sequence() == 6ul);
            REQUIRE(peer_folder_info->get_file_infos().size() == 1);
            auto peer_file = peer_folder_info->get_file_infos().by_name("a.txt");
            REQUIRE(peer_file);

            auto &unknown_folders = cluster->get_pending_folders();
            CHECK(std::distance(unknown_folders.begin(), unknown_folders.end()) == 1);

            sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
            sup->do_process();
            REQUIRE(reply);
            REQUIRE(!reply->payload.ee);

            auto cluster_clone = make_cluster();
            {
                REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                REQUIRE(cluster_clone->get_blocks().size() == 1);
                CHECK(cluster_clone->get_blocks().by_hash(b_hash));
                auto folder = cluster_clone->get_folders().by_id(folder_id);
                auto peer_folder_info = folder->get_folder_infos().by_device(*peer_device);
                REQUIRE(peer_folder_info);
                REQUIRE(peer_folder_info->get_file_infos().size() == 1);
                REQUIRE(peer_folder_info->get_file_infos().by_name("a.txt"));
                REQUIRE(cluster_clone->get_pending_folders().size() == 1);
            }

            auto pr_msg = proto::ClusterConfig();
            auto &pr_f = proto::add_folders(pr_msg);
            proto::set_id(pr_f, folder_id);
            auto &pr_device = proto::add_devices(pr_f);
            proto::set_id(pr_device, peer_device->device_id().get_sha256());
            proto::set_max_sequence(pr_device, 1);
            proto::set_index_id(pr_device, peer_folder_info->get_index() + 1);
            auto diff =
                diff::peer::cluster_update_t::create({}, *cluster, *sup->sequencer, *peer_device, pr_msg).value();

            sup->send<model::payload::model_update_t>(sup->get_address(), diff, nullptr);
            sup->do_process();

            sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
            sup->do_process();
            REQUIRE(reply);
            REQUIRE(!reply->payload.ee);

            cluster_clone = make_cluster();
            {
                REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                REQUIRE(cluster_clone->get_blocks().size() == 0);
                auto &fis = cluster_clone->get_folders().by_id(folder_id)->get_folder_infos();
                REQUIRE(fis.size() == 2);
                auto folder_info = fis.by_device(*peer_device);
                REQUIRE(folder_info);
                REQUIRE(fis.by_device(*cluster->get_device()));
                REQUIRE(cluster_clone->get_pending_folders().size() == 0);
            }
        }
    };
    F().run();
}

void test_unshare_and_remove_folder() {
    struct F : fixture_t {
        void main() noexcept override {
            auto sha256 = peer_device->device_id().get_sha256();
            auto folder_id = "1234-5678";

            auto file = proto::FileInfo();
            proto::set_name(file, "a.txt");
            proto::set_size(file, 5ul);
            proto::set_block_size(file, 5ul);
            proto::set_sequence(file, 6ul);

            auto b_hash = utils::sha256_digest(as_bytes("12345")).value();
            auto &b = proto::add_blocks(file);
            proto::set_size(b, 5ul);
            proto::set_hash(b, b_hash);

            auto builder = diff_builder_t(*cluster);
            builder.update_peer(peer_device->device_id())
                .apply(*sup)
                .upsert_folder(folder_id, "/my/path")
                .apply(*sup)
                .configure_cluster(sha256)
                .add(sha256, folder_id, 5, proto::get_sequence(file))
                .finish()
                .share_folder(sha256, folder_id)
                .apply(*sup)
                .make_index(sha256, folder_id)
                .add(file, peer_device)
                .finish()
                .apply(*sup);

            REQUIRE(cluster->get_blocks().size() == 1);
            auto block = cluster->get_blocks().by_hash(b_hash);
            REQUIRE(block);

            auto folder = cluster->get_folders().by_id(folder_id);
            auto peer_folder_info = folder->get_folder_infos().by_device(*peer_device);

            REQUIRE(peer_folder_info);
            CHECK(peer_folder_info->get_max_sequence() == 6ul);
            REQUIRE(peer_folder_info->get_file_infos().size() == 1);
            auto peer_file = peer_folder_info->get_file_infos().by_name("a.txt");
            REQUIRE(peer_file);

            SECTION("unshare") {
                builder.unshare_folder(*peer_folder_info).apply(*sup);

                sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                sup->do_process();
                REQUIRE(reply);
                REQUIRE(!reply->payload.ee);

                auto cluster_clone = make_cluster();
                {
                    REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                    auto &fis = cluster_clone->get_folders().by_id(folder_id)->get_folder_infos();
                    REQUIRE(fis.size() == 1);
                    REQUIRE(!fis.by_device(*peer_device));
                    REQUIRE(fis.by_device(*cluster->get_device()));
                    REQUIRE(cluster_clone->get_blocks().size() == 0);
                }
            }

            SECTION("remove") {
                builder.remove_folder(*folder).apply(*sup);

                sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                sup->do_process();
                REQUIRE(reply);
                REQUIRE(!reply->payload.ee);

                auto cluster_clone = make_cluster();
                {
                    REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                    REQUIRE(cluster_clone->get_folders().size() == 0);
                    REQUIRE(cluster_clone->get_blocks().size() == 0);
                }
            }
        }
    };
    F().run();
}

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

            auto sha256 = peer_device->device_id().get_sha256();
            auto folder_id = "1234-5678";

            auto file = proto::FileInfo();
            proto::set_name(file, "a.txt");
            proto::set_sequence(file, 6ul);

            auto &v = proto::get_version(file);
            proto::add_counters(v, proto::Counter(peer_device->device_id().get_uint(), 1));

            auto builder = diff_builder_t(*cluster);
            builder.update_peer(peer_device->device_id())
                .apply(*sup)
                .upsert_folder(folder_id, "/my/path")
                .configure_cluster(sha256)
                .add(sha256, folder_id, 5, proto::get_sequence(file))
                .finish()
                .apply(*sup)
                .share_folder(sha256, folder_id)
                .apply(*sup);

            auto folder = cluster->get_folders().by_id(folder_id);
            auto folder_my = folder->get_folder_infos().by_device(*my_device);
            auto folder_peer = folder->get_folder_infos().by_device(*peer_device);

            SECTION("file without blocks") {
                builder.make_index(sha256, folder_id).add(file, peer_device).finish().apply(*sup);

                auto file_peer = folder_peer->get_file_infos().by_name(proto::get_name(file));
                REQUIRE(file_peer);
                builder.remote_copy(*file_peer).apply(*sup);

                sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                sup->do_process();
                REQUIRE(reply);
                REQUIRE(!reply->payload.ee);

                auto cluster_clone = make_cluster();
                {
                    REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                    REQUIRE(cluster_clone->get_blocks().size() == 0);
                    auto &fis = cluster_clone->get_folders().by_id(folder_id)->get_folder_infos();
                    REQUIRE(fis.size() == 2);
                    auto folder_info_clone = fis.by_device(*cluster_clone->get_device());
                    auto file_clone = folder_info_clone->get_file_infos().by_name(proto::get_name(file));
                    REQUIRE(file_clone);
                    REQUIRE(file_clone->get_name() == proto::get_name(file));
                    REQUIRE(file_clone->get_blocks().size() == 0);
                    REQUIRE(file_clone->get_sequence() == 1);
                    REQUIRE(folder_info_clone->get_max_sequence() == 1);
                }
            }

            SECTION("file with blocks") {
                proto::set_size(file, 5ul);
                proto::set_block_size(file, 5ul);

                auto b_hash = utils::sha256_digest(as_bytes("12345")).value();
                auto &b = proto::add_blocks(file);
                proto::set_size(b, 5ul);
                proto::set_hash(b, b_hash);

                builder.make_index(sha256, folder_id).add(file, peer_device).finish().apply(*sup);

                auto folder = cluster->get_folders().by_id(folder_id);
                auto folder_my = folder->get_folder_infos().by_device(*my_device);
                auto folder_peer = folder->get_folder_infos().by_device(*peer_device);
                auto file_peer = folder_peer->get_file_infos().by_name(proto::get_name(file));
                REQUIRE(file_peer);

                builder.remote_copy(*file_peer).apply(*sup);
                REQUIRE(folder_my->get_max_sequence() == 1);

                {
                    sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                    sup->do_process();
                    REQUIRE(reply);
                    REQUIRE(!reply->payload.ee);

                    auto cluster_clone = make_cluster();
                    REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                    REQUIRE(cluster_clone->get_blocks().size() == 1);
                    auto &fis = cluster_clone->get_folders().by_id(folder_id)->get_folder_infos();
                    REQUIRE(fis.size() == 2);
                    auto folder_info_clone = fis.by_device(*cluster_clone->get_device());
                    auto file_clone = folder_info_clone->get_file_infos().by_name(proto::get_name(file));
                    REQUIRE(file_clone);
                    REQUIRE(file_clone->get_name() == proto::get_name(file));
                    REQUIRE(file_clone->get_blocks().size() == 1);
                    REQUIRE(file_clone->get_sequence() == 1);
                    REQUIRE(folder_info_clone->get_max_sequence() == 1);
                }

                file_peer = folder_peer->get_file_infos().by_name(proto::get_name(file));
                file_peer->mark_local_available(0);
                REQUIRE(file_peer->is_locally_available());

                builder.remote_copy(*file_peer).apply(*sup);

                {
                    sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                    sup->do_process();
                    REQUIRE(reply);
                    REQUIRE(!reply->payload.ee);

                    auto cluster_clone = make_cluster();
                    REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                    REQUIRE(cluster_clone->get_blocks().size() == 1);
                    auto &fis = cluster_clone->get_folders().by_id(folder_id)->get_folder_infos();
                    REQUIRE(fis.size() == 2);
                    auto folder_info_clone = fis.by_device(*cluster_clone->get_device());
                    auto file_clone = folder_info_clone->get_file_infos().by_name(proto::get_name(file));
                    REQUIRE(file_clone);
                    REQUIRE(file_clone->get_name() == proto::get_name(file));
                    REQUIRE(file_clone->get_blocks().size() == 1);
                    REQUIRE(file_clone->get_blocks().at(0));
                    REQUIRE(file_clone->get_sequence() == 2);
                    REQUIRE(folder_info_clone->get_max_sequence() == 2);
                }
            }
        }
    };
    F().run();
}

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

            auto folder_id = "1234-5678";

            auto pr_file = proto::FileInfo();
            proto::set_name(pr_file, "a.txt");
            proto::set_size(pr_file, 5ul);

            auto hash = utils::sha256_digest(as_bytes("12345")).value();
            auto &pr_block = proto::add_blocks(pr_file);
            proto::set_size(pr_block, 5ul);
            proto::set_hash(pr_block, hash);

            auto builder = diff_builder_t(*cluster);
            builder.upsert_folder(folder_id, "/my/path").apply(*sup).local_update(folder_id, pr_file).apply(*sup);

            SECTION("check saved file with new blocks") {
                sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                sup->do_process();
                REQUIRE(reply);
                REQUIRE(!reply->payload.ee);
                auto cluster_clone = make_cluster();
                REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));

                auto folder = cluster_clone->get_folders().by_id(folder_id);
                auto folder_my = folder->get_folder_infos().by_device(*my_device);
                auto file = folder_my->get_file_infos().by_name("a.txt");
                REQUIRE(file);
                CHECK(cluster_clone->get_blocks().size() == 1);
                CHECK(file->get_blocks().size() == 1);
            }

            proto::set_deleted(pr_file, true);
            proto::set_size(pr_file, 0);
            proto::clear_blocks(pr_file);
            builder.local_update(folder_id, pr_file).apply(*sup);

            SECTION("check deleted blocks") {
                sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                sup->do_process();
                REQUIRE(reply);
                REQUIRE(!reply->payload.ee);
                auto cluster_clone = make_cluster();
                REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));

                auto folder = cluster_clone->get_folders().by_id(folder_id);
                auto folder_my = folder->get_folder_infos().by_device(*my_device);
                auto file = folder_my->get_file_infos().by_name("a.txt");
                REQUIRE(file);
                CHECK(file->is_deleted());
                CHECK(cluster_clone->get_blocks().size() == 0);
                CHECK(file->get_blocks().size() == 0);
            }
        }
    };
    F().run();
};

void test_peer_going_offline() {
    struct F : fixture_t {
        void main() noexcept override {
            auto builder = diff_builder_t(*cluster);
            builder.update_peer(peer_device->device_id()).apply(*sup);

            auto sha256 = peer_device->device_id().get_sha256();
            db::Device db_peer;
            auto peer = cluster->get_devices().by_sha256(sha256);
            REQUIRE(db::get_last_seen(db_peer) == 0);
            auto connection_id = std::string("tcp://1.1.1.1:1");
            peer->update_state(device_state_t::online, connection_id);

            builder.update_state(*peer, {}, device_state_t::offline, connection_id).apply(*sup);

            sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
            sup->do_process();
            REQUIRE(reply);

            auto cluster_clone = make_cluster();
            REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
            auto peer_clone = cluster_clone->get_devices().by_sha256(sha256);
            db::Device db_peer_clone;
            peer_clone->serialize(db_peer_clone);
            CHECK((peer->get_last_seen() - peer_clone->get_last_seen()).total_seconds() < 2);
            REQUIRE(db::get_last_seen(db_peer_clone) != 0);
        }
    };
    F().run();
};

void test_remove_peer() {
    struct F : fixture_t {
        void main() noexcept override {
            auto sha256 = peer_device->device_id().get_sha256();
            auto folder_id = "1234-5678";
            auto unknown_folder_id = "5678-999";

            auto file = proto::FileInfo();
            proto::set_name(file, "a.txt");
            proto::set_size(file, 5ul);
            proto::set_block_size(file, 5ul);
            proto::set_sequence(file, 6ul);

            auto b_hash = utils::sha256_digest(as_bytes("12345")).value();
            auto &b = proto::add_blocks(file);
            proto::set_size(b, 5ul);
            proto::set_hash(b, b_hash);

            auto builder = diff_builder_t(*cluster);
            builder.update_peer(peer_device->device_id())
                .apply(*sup)
                .upsert_folder(folder_id, "/my/path")
                .configure_cluster(sha256)
                .add(sha256, folder_id, 5, proto::get_sequence(file))
                .add(sha256, unknown_folder_id, 5, 5)
                .finish()
                .apply(*sup)
                .share_folder(sha256, folder_id)
                .apply(*sup)
                .make_index(sha256, folder_id)
                .add(file, peer_device)
                .finish()
                .apply(*sup);

            CHECK(cluster->get_pending_folders().size() == 1);

            REQUIRE(cluster->get_blocks().size() == 1);
            auto block = cluster->get_blocks().by_hash(b_hash);
            REQUIRE(block);

            auto folder = cluster->get_folders().by_id(folder_id);
            auto peer_folder_info = folder->get_folder_infos().by_device(*peer_device);

            REQUIRE(peer_folder_info);
            CHECK(peer_folder_info->get_max_sequence() == 6ul);
            REQUIRE(peer_folder_info->get_file_infos().size() == 1);
            auto peer_file = peer_folder_info->get_file_infos().by_name("a.txt");
            REQUIRE(peer_file);

            builder.remove_peer(*peer_device).apply(*sup);

            sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
            sup->do_process();
            REQUIRE(reply);
            REQUIRE(!reply->payload.ee);

            auto cluster_clone = make_cluster(false);
            {
                REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                CHECK(cluster_clone->get_pending_folders().size() == 0);
                CHECK(cluster_clone->get_devices().size() == 1);
                REQUIRE(cluster_clone->get_blocks().size() == 0);
            }
        }
    };
    F().run();
}

void test_update_peer() {
    struct F : fixture_t {
        void main() noexcept override {
            auto builder = diff_builder_t(*cluster);

            db::SomeDevice db;
            db::set_name(db, "x1");

            SECTION("unknown device is removed") {
                builder.add_unknown_device(peer_device->device_id(), db)
                    .apply(*sup)
                    .update_peer(peer_device->device_id(), "p1")
                    .apply(*sup);
            }

            SECTION("ignored device is removed") {
                builder.add_ignored_device(peer_device->device_id(), db)
                    .apply(*sup)
                    .update_peer(peer_device->device_id(), "p1")
                    .apply(*sup);
            }

            {
                sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                sup->do_process();
                REQUIRE(reply);
                REQUIRE(!reply->payload.ee);
                auto cluster_clone = make_cluster();
                REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                CHECK(cluster_clone->get_pending_devices().size() == 0);
                CHECK(cluster_clone->get_ignored_devices().size() == 0);
                CHECK(cluster_clone->get_devices().size() == 2);
            }
        }
    };
    F().run();
}

void test_peer_3_folders_6_files() {
    struct F : fixture_t {
        void main() noexcept override {
            auto sha256 = peer_device->device_id().get_sha256();
            auto f1_id = "123";
            auto f2_id = "356";
            auto f3_id = "789";

            auto next_sequence = 6ul;

            auto make_file = [&](std::string name) {
                auto file = proto::FileInfo();
                proto::set_name(file, name);
                proto::set_sequence(file, ++next_sequence);
                auto &v = proto::get_version(file);
                proto::add_counters(v, proto::Counter(peer_device->device_id().get_uint(), 1));
                return file;
            };

            auto builder = diff_builder_t(*cluster);

            // clang-format off
            builder
                .update_peer(peer_device->device_id())
                .apply(*sup)
                .upsert_folder(f1_id, "/my/path1", "my-label1")
                .upsert_folder(f2_id, "/my/path2", "my-label2")
                .upsert_folder(f3_id, "/my/path3", "my-label3")
                .configure_cluster(sha256)
                    .add(sha256, f1_id, 10, 5)
                    .add(sha256, f2_id, 11, 5)
                    .add(sha256, f3_id, 12, 5)
                .finish()
                .apply(*sup)
                    .share_folder(sha256, f1_id)
                    .share_folder(sha256, f2_id)
                    .share_folder(sha256, f3_id)
                .apply(*sup)
                    .make_index(sha256, f1_id).add(make_file("f1.1"), peer_device).add(make_file("f1.2"), peer_device).finish()
                    .make_index(sha256, f2_id).add(make_file("f2.1"), peer_device).add(make_file("f2.2"), peer_device).finish()
                    .make_index(sha256, f3_id).add(make_file("f3.1"), peer_device).add(make_file("f3.2"), peer_device).finish()
                .apply(*sup);
            // clang-format on

            {
                auto get_peer_file = [&](std::string_view folder_id, std::string_view name) {
                    auto folder = cluster->get_folders().by_id(folder_id);
                    auto folder_info = folder->get_folder_infos().by_device(*peer_device);
                    auto file = folder_info->get_file_infos().by_name(name);
                    return file;
                };

                auto file_11 = get_peer_file(f1_id, "f1.1");
                auto file_12 = get_peer_file(f1_id, "f1.2");
                auto file_21 = get_peer_file(f2_id, "f2.1");
                auto file_22 = get_peer_file(f2_id, "f2.2");
                auto file_31 = get_peer_file(f3_id, "f3.1");
                auto file_32 = get_peer_file(f3_id, "f3.2");

                REQUIRE(file_11);
                REQUIRE(file_12);
                REQUIRE(file_21);
                REQUIRE(file_22);
                REQUIRE(file_31);
                REQUIRE(file_32);

                // clang-format off
                builder
                    .remote_copy(*file_11).remote_copy(*file_12)
                    .remote_copy(*file_21).remote_copy(*file_22)
                    .remote_copy(*file_31).remote_copy(*file_32)
                .apply(*sup);
                // clang-format on
            }

            {
                sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
                sup->do_process();
                REQUIRE(reply);
                REQUIRE(!reply->payload.ee);
                auto cluster_clone = make_cluster();
                REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));

                auto get_my_file = [&](std::string_view folder_id, std::string_view name) {
                    auto folder = cluster->get_folders().by_id(folder_id);
                    auto folder_info = folder->get_folder_infos().by_device(*my_device);
                    auto file = folder_info->get_file_infos().by_name(name);
                    return file;
                };

                auto file_11 = get_my_file(f1_id, "f1.1");
                auto file_12 = get_my_file(f1_id, "f1.2");
                auto file_21 = get_my_file(f2_id, "f2.1");
                auto file_22 = get_my_file(f2_id, "f2.2");
                auto file_31 = get_my_file(f3_id, "f3.1");
                auto file_32 = get_my_file(f3_id, "f3.2");

                REQUIRE(file_11);
                REQUIRE(file_12);
                REQUIRE(file_21);
                REQUIRE(file_22);
                REQUIRE(file_31);
                REQUIRE(file_32);
            }
        }
    };
    F().run();
}

void test_db_migration_1_2() {
    static constexpr auto folder_id = "1234-5678";

    struct F1 : fixture_t {
        void main() noexcept override {

            auto sha256 = peer_device->device_id().get_sha256();
            auto builder = diff_builder_t(*cluster);
            builder.update_peer(peer_device->device_id())
                .apply(*sup)
                .upsert_folder(folder_id, "/my/path")
                .apply(*sup)
                .share_folder(sha256, folder_id)
                .apply(*sup);

            auto &folder_infos = cluster->get_folders().by_id(folder_id)->get_folder_infos();
            auto fi_my = folder_infos.by_device(*my_device);
            auto fi_peer = folder_infos.by_device(*peer_device);

            REQUIRE(fi_my->is_introduced_by(my_device->device_id()));
            REQUIRE(fi_peer->is_introduced_by(my_device->device_id()));

            auto &db_env = db_actor->access<env>();
            auto txn_opt = db::make_transaction(db::transaction_type_t::RW, db_env);
            REQUIRE(txn_opt);
            auto &txn = txn_opt.value();
            REQUIRE(db::save_version(1, txn));

            auto save = [&](const model::folder_info_t &fi) -> outcome::result<void> {
                auto fi_db = db::FolderInfo();
                auto fi_key = fi.get_key();
                fi.serialize(fi_db);
                db::set_introducer_device_key(fi_db, {});
                auto fi_data = db::encode(fi_db);
                return db::save({fi_key, fi_data}, txn);
            };

            REQUIRE(save(*fi_my));
            REQUIRE(save(*fi_peer));
            REQUIRE(txn.commit());
        }
    };
    struct F2 : fixture_t {
        F2(fixture_t &other) : fixture_t(std::move(other)) {}
        void main() noexcept override {
            sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
            sup->do_process();
            REQUIRE(reply);
            REQUIRE(!reply->payload.ee);
            auto cluster_clone = make_cluster();
            REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));

            auto f_cloned = cluster_clone->get_folders().by_id(folder_id);
            auto &folder_infos = f_cloned->get_folder_infos();
            auto fi_my = folder_infos.by_device(*my_device);
            auto fi_peer = folder_infos.by_device(*peer_device);

            CHECK(fi_my->is_introduced_by(my_device->device_id()));
            CHECK(fi_peer->is_introduced_by(my_device->device_id()));
        }
    };
    F2(F1().run()).run();
}

void test_corrupted_file() {
    struct F : fixture_t {
        void main() noexcept override {
            auto folder_id = "1234-5678";
            auto builder = diff_builder_t(*cluster);
            builder.upsert_folder(folder_id, "/my/path").apply(*sup);
            CHECK(static_cast<r::actor_base_t *>(db_actor.get())->access<to::state>() == r::state_t::OPERATIONAL);

            auto pr_file = proto::FileInfo();
            proto::set_name(pr_file, "a.txt");
            proto::set_size(pr_file, 5ul);

            auto hash = utils::sha256_digest(as_bytes("12345")).value();
            auto &pr_block = proto::add_blocks(pr_file);
            proto::set_size(pr_block, 5ul);
            proto::set_hash(pr_block, hash);

            builder.local_update(folder_id, pr_file).apply(*sup);
            auto &block = *cluster->get_blocks().begin()->item;

            auto folder = cluster->get_folders().by_id(folder_id);
            auto fi = folder->get_folder_infos().by_device(*my_device);
            ;
            auto file = fi->get_file_infos().by_name("a.txt");
            REQUIRE(file);

            auto &db_env = db_actor->access<env>();
            auto txn_opt = db::make_transaction(db::transaction_type_t::RW, db_env);
            REQUIRE(txn_opt);
            auto &txn = txn_opt.value();
            auto key = block.get_key();
            REQUIRE(db::remove(key, txn));
            REQUIRE(!txn.commit().has_error());

            sup->request<net::payload::load_cluster_request_t>(db_addr).send(timeout);
            sup->do_process();
            REQUIRE(reply);
            REQUIRE(!reply->payload.ee);

            auto cluster_clone = make_cluster(false);
            {
                REQUIRE(reply->payload.res.diff->apply(*cluster_clone, get_apply_controller()));
                auto &folder_infos = cluster_clone->get_folders().by_id(folder_id)->get_folder_infos();
                auto folder_info = folder_infos.by_device(*my_device);
                CHECK(folder_info->get_file_infos().size() == 0);
                auto &blocks = cluster_clone->get_blocks();
                CHECK(blocks.size() == 0);
            }
        }
    };

    F().run();
}

int _init() {
    REGISTER_TEST_CASE(test_db_population, "test_db_population", "[db]");
    REGISTER_TEST_CASE(test_loading_empty_db, "test_loading_empty_db", "[db]");
    REGISTER_TEST_CASE(test_unknown_and_ignored_devices_1, "test_unknown_and_ignored_devices_1", "[db]");
    REGISTER_TEST_CASE(test_unknown_and_ignored_devices_2, "test_unknown_and_ignored_devices_2", "[db]");
    REGISTER_TEST_CASE(test_folder_upserting, "test_folder_upserting", "[db]");
    REGISTER_TEST_CASE(test_peer_updating, "test_peer_updating", "[db]");
    REGISTER_TEST_CASE(test_folder_sharing, "test_folder_sharing", "[db]");
    REGISTER_TEST_CASE(test_cluster_update_and_remove, "test_cluster_update_and_remove", "[db]");
    REGISTER_TEST_CASE(test_unshare_and_remove_folder, "test_unshare_and_remove_folder", "[db]");
    REGISTER_TEST_CASE(test_remote_copy, "test_remote_copy", "[db]");
    REGISTER_TEST_CASE(test_local_update, "test_local_update", "[db]");
    REGISTER_TEST_CASE(test_peer_going_offline, "test_peer_going_offline", "[db]");
    REGISTER_TEST_CASE(test_remove_peer, "test_remove_peer", "[db]");
    REGISTER_TEST_CASE(test_update_peer, "test_update_peer", "[db]");
    REGISTER_TEST_CASE(test_peer_3_folders_6_files, "test_peer_3_folders_6_files", "[db]");
    REGISTER_TEST_CASE(test_db_migration_1_2, "test_db_migration_1_2", "[db]");
    REGISTER_TEST_CASE(test_corrupted_file, "test_corrupted_file", "[db]");
    return 1;
}

static int v = _init();

repo_name: basiliscos/syncspirit

license: mit


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

#include "test-utils.h"
#include "fs/file_actor.h"
#include "fs/utils.h"
#include "diff-builder.h"
#include "net/messages.h"
#include "test_supervisor.h"
#include "access.h"
#include "model/cluster.h"
#include "model/misc/resolver.h"
#include "access.h"
#include <filesystem>

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

namespace bfs = std::filesystem;

namespace {

struct fixture_t {
    using msg_t = net::message::load_cluster_response_t;
    using msg_ptr_t = r::intrusive_ptr_t<msg_t>;
    using blk_res_t = fs::message::block_response_t;
    using blk_res_ptr_t = r::intrusive_ptr_t<blk_res_t>;

    fixture_t() noexcept : root_path{unique_path()}, path_guard{root_path} {
        test::init_logging();
        bfs::create_directory(root_path);
        sequencer = make_sequencer(67);
    }

    virtual supervisor_t::configure_callback_t configure() noexcept {
        return [&](r::plugin::plugin_base_t &plugin) {
            plugin.template with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
                p.subscribe_actor(r::lambda<msg_t>([&](msg_t &msg) { reply = &msg; }));
                p.subscribe_actor(r::lambda<blk_res_t>([&](blk_res_t &msg) { block_reply = &msg; }));
            });
        };
    }

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

        cluster = new cluster_t(my_device, 1);
        auto peer_id =
            device_id_t::from_string("VUV42CZ-IQD5A37-RPEBPM4-VVQK6E4-6WSKC7B-PVJQHHD-4PZD44V-ENC6WAZ").value();
        peer_device = device_t::create(peer_id, "peer-device").value();

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

        r::system_context_t ctx;
        sup = ctx.create_supervisor<supervisor_t>()
                  .auto_finish(false)
                  .auto_ack_blocks(false)
                  .timeout(timeout)
                  .create_registry()
                  .finish();
        sup->cluster = cluster;
        sup->configure_callback = configure();

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

        auto sha256 = peer_device->device_id().get_sha256();
        file_actor = sup->create_actor<fs::file_actor_t>()
                         .mru_size(2)
                         .cluster(cluster)
                         .sequencer(sup->sequencer)
                         .timeout(timeout)
                         .finish();
        sup->do_process();
        CHECK(static_cast<r::actor_base_t *>(file_actor.get())->access<to::state>() == r::state_t::OPERATIONAL);
        file_addr = file_actor->get_address();

        auto builder = diff_builder_t(*cluster);
        builder.upsert_folder(folder_id, root_path.string(), "my-label")
            .apply(*sup)
            .update_peer(peer_device->device_id(), "some_name", "some-cn", true)
            .apply(*sup)
            .share_folder(sha256, folder_id)
            .apply(*sup);

        folder = cluster->get_folders().by_id(folder_id);
        folder_my = folder->get_folder_infos().by_device(*my_device);
        folder_peer = folder->get_folder_infos().by_device(*peer_device);

        main();
        reply.reset();

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

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

    virtual void main() noexcept {}

    r::address_ptr_t file_addr;
    r::pt::time_duration timeout = r::pt::millisec{10};
    cluster_ptr_t cluster;
    model::sequencer_ptr_t sequencer;
    model::device_ptr_t peer_device;
    model::device_ptr_t my_device;
    model::folder_ptr_t folder;
    model::folder_info_ptr_t folder_my;
    model::folder_info_ptr_t folder_peer;
    r::intrusive_ptr_t<supervisor_t> sup;
    r::intrusive_ptr_t<fs::file_actor_t> file_actor;
    bfs::path root_path;
    path_guard_t path_guard;
    r::system_context_t ctx;
    msg_ptr_t reply;
    blk_res_ptr_t block_reply;
    std::string_view folder_id = "1234-5678";
};
} // namespace

void test_remote_copy() {
    struct F : fixture_t {
        void main() noexcept override {
            proto::FileInfo pr_fi;
            std::int64_t modified = 1641828421;
            proto::set_name(pr_fi, "q.txt");
            proto::set_modified_s(pr_fi, modified);
            proto::set_sequence(pr_fi, folder_peer->get_max_sequence() + 1);

            auto &v = proto::get_version(pr_fi);
            proto::add_counters(v, proto::Counter(peer_device->device_id().get_uint(), 1));

            auto builder = diff_builder_t(*cluster, file_addr);

            auto make_file = [&]() {
                auto file = file_info_t::create(sequencer->next_uuid(), pr_fi, folder_peer).value();
                REQUIRE(folder_peer->add_strict(file));
                return file;
            };

            SECTION("empty regular file") {
                auto peer_file = make_file();
                builder.remote_copy(*peer_file).apply(*sup);

                auto my_file = folder_my->get_file_infos().by_name(peer_file->get_name());
                auto &path = my_file->get_path();
                REQUIRE(bfs::exists(path));
                REQUIRE(bfs::file_size(path) == 0);
                REQUIRE(to_unix(bfs::last_write_time(path)) == 1641828421);
            }

            SECTION("empty regular file a subdir") {
                auto name = std::string_view("a/b/c/d/e.txt");
                proto::set_name(pr_fi, name);
                auto peer_file = make_file();
                builder.remote_copy(*peer_file).apply(*sup);

                auto file = folder_my->get_file_infos().by_name(name);
                auto &path = file->get_path();
                REQUIRE(bfs::exists(path));
                REQUIRE(bfs::file_size(path) == 0);
            }

            SECTION("non-empty regular file") {
                proto::set_size(pr_fi, 5);
                proto::set_block_size(pr_fi, 5);

                auto block_hash = utils::sha256_digest(as_bytes("12345")).value();
                auto &b = proto::add_blocks(pr_fi);
                proto::set_hash(b, block_hash);
                proto::set_size(b, 5);
                auto bi = block_info_t::create(b).value();
                auto &blocks_map = cluster->get_blocks();
                blocks_map.put(bi);

                auto peer_file = make_file();
                peer_file->assign_block(bi, 0);
                builder.remote_copy(*peer_file).apply(*sup);

                auto file = folder_my->get_file_infos().by_name(proto::get_name(pr_fi));

                auto filename = std::string(file->get_name()) + ".syncspirit-tmp";
                auto path = root_path / filename;
                REQUIRE(bfs::exists(path));
                REQUIRE(bfs::file_size(path) == 5);
            }

            SECTION("directory") {
                proto::set_type(pr_fi, proto::FileInfoType::DIRECTORY);

                auto peer_file = make_file();
                builder.remote_copy(*peer_file).apply(*sup);

                auto file = folder_my->get_file_infos().by_name(proto::get_name(pr_fi));

                auto &path = file->get_path();
                REQUIRE(bfs::exists(path));
                REQUIRE(bfs::is_directory(path));
            }

            SECTION("symlink") {
                bfs::path target = root_path / "some-existing-file";
                write_file(target, "zzz");

                proto::set_type(pr_fi, proto::FileInfoType::SYMLINK);
                proto::set_symlink_target(pr_fi, target.string());

                auto peer_file = make_file();
                builder.remote_copy(*peer_file).apply(*sup);

                auto file = folder_my->get_file_infos().by_name(proto::get_name(pr_fi));

                auto &path = file->get_path();
                CHECK(!bfs::exists(path));
#ifndef SYNCSPIRIT_WIN
                CHECK(bfs::is_symlink(path));
                CHECK(bfs::read_symlink(path) == target);
#endif
            }

            SECTION("deleted file") {
                proto::set_deleted(pr_fi, true);
                bfs::path target = root_path / proto::get_name(pr_fi);
                write_file(target, "zzz");
                REQUIRE(bfs::exists(target));

                auto peer_file = make_file();
                builder.remote_copy(*peer_file).apply(*sup);

                auto file = folder_my->get_file_infos().by_name(proto::get_name(pr_fi));
                CHECK(file->is_deleted());

                REQUIRE(!bfs::exists(target));
            }
        }
    };
    F().run();
}

void test_append_block() {
    struct F : fixture_t {
        void main() noexcept override {
            std::int64_t modified = 1641828421;
            proto::FileInfo pr_source;
            auto next_sequence = 7ul;

            proto::set_name(pr_source, "q.txt");
            proto::set_modified_s(pr_source, modified);
            proto::set_block_size(pr_source, 5);

            auto &v = proto::get_version(pr_source);
            proto::add_counters(v, proto::Counter(peer_device->device_id().get_uint(), 1));

            auto block_hash_1 = utils::sha256_digest(as_bytes("12345")).value();
            auto bi = proto::BlockInfo();
            proto::set_hash(bi, block_hash_1);
            proto::set_size(bi, 5);
            auto b = block_info_t::create(bi).value();

            auto block_hash_2 = utils::sha256_digest(as_bytes("67890")).value();
            auto bi2 = proto::BlockInfo();
            proto::set_hash(bi2, block_hash_2);
            proto::set_size(bi2, 5);
            auto b2 = block_info_t::create(bi).value();

            cluster->get_blocks().put(b);
            cluster->get_blocks().put(b2);
            auto blocks = std::vector<block_info_ptr_t>{b, b2};

            auto make_file = [&](size_t block_count) {
                proto::set_sequence(pr_source, ++next_sequence);

                auto copy = pr_source;
                auto &v2 = proto::get_version(copy);
                auto version = model::version_t(v2);
                version.update(*peer_device);
                version.to_proto(v2);

                auto file = file_info_t::create(sequencer->next_uuid(), copy, folder_peer).value();
                for (size_t i = 0; i < block_count; ++i) {
                    file->assign_block(blocks[i], i);
                }
                REQUIRE(folder_peer->add_strict(file));
                return file;
            };

            auto builder = diff_builder_t(*cluster, file_addr);

            SECTION("file with 1 block") {
                proto::set_size(pr_source, 5);
                auto peer_file = make_file(1);
                builder.append_block(*peer_file, 0, as_owned_bytes("12345"))
                    .apply(*sup)
                    .finish_file(*peer_file)
                    .apply(*sup);

                auto file = folder_my->get_file_infos().by_name(proto::get_name(pr_source));
                auto path = root_path / std::string(file->get_name());
                REQUIRE(bfs::exists(path));
                REQUIRE(bfs::file_size(path) == 5);
                auto data = read_file(path);
                CHECK(data == "12345");
                CHECK(to_unix(bfs::last_write_time(path)) == 1641828421);
            }
            SECTION("file with 2 different blocks") {
                proto::set_size(pr_source, 10);

                auto peer_file = make_file(2);

                builder.append_block(*peer_file, 0, as_owned_bytes("12345")).apply(*sup);

                auto filename = std::string(peer_file->get_name()) + ".syncspirit-tmp";
                auto path = root_path / filename;
#ifndef SYNCSPIRIT_WIN
                REQUIRE(bfs::exists(path));
                REQUIRE(bfs::file_size(path) == 10);
                auto data = read_file(path);
                CHECK(data.substr(0, 5) == "12345");
#endif
                builder.append_block(*peer_file, 1, as_owned_bytes("67890")).apply(*sup);

                SECTION("add 2nd block") {
                    builder.finish_file(*peer_file).apply(*sup);

                    filename = std::string(peer_file->get_name());
                    path = root_path / filename;
                    REQUIRE(bfs::exists(path));
                    REQUIRE(bfs::file_size(path) == 10);
                    auto data = read_file(path);
                    CHECK(data == "1234567890");
                    CHECK(to_unix(bfs::last_write_time(path)) == 1641828421);
                }

#ifndef SYNCSPIRIT_WIN
                SECTION("remove folder (simulate err)") {
                    bfs::remove_all(root_path);
                    builder.finish_file(*peer_file).apply(*sup);
                    CHECK(static_cast<r::actor_base_t *>(file_actor.get())->access<to::state>() ==
                          r::state_t::SHUT_DOWN);
                }
#endif
            }
        }
    };
    F().run();
}

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

            auto hash_1 = utils::sha256_digest(as_bytes("12345")).value();
            auto bi = proto::BlockInfo();
            proto::set_size(bi, 5);
            proto::set_hash(bi, hash_1);
            auto b = block_info_t::create(bi).value();

            auto hash_2 = utils::sha256_digest(as_bytes("67890")).value();
            auto bi2 = proto::BlockInfo();
            proto::set_size(bi2, 5);
            proto::set_hash(bi2, hash_2);
            auto b2 = block_info_t::create(bi2).value();

            cluster->get_blocks().put(b);
            cluster->get_blocks().put(b2);
            auto blocks = std::vector<block_info_ptr_t>{b, b2};

            std::int64_t modified = 1641828421;
            proto::FileInfo pr_source;
            proto::set_name(pr_source, "a.txt");
            proto::set_modified_s(pr_source, modified);
            proto::set_block_size(pr_source, 5);

            auto &v = proto::get_version(pr_source);
            proto::add_counters(v, proto::Counter(peer_device->device_id().get_uint(), 1));

            auto next_sequence = 7ul;

            auto make_file = [&](const proto::FileInfo &fi, size_t count) {
                auto copy = fi;
                proto::set_sequence(copy, ++next_sequence);
                auto file = file_info_t::create(sequencer->next_uuid(), copy, folder_peer).value();
                for (size_t i = 0; i < count; ++i) {
                    file->assign_block(blocks[i], i);
                }
                REQUIRE(folder_peer->add_strict(file));
                return file;
            };

            auto builder = diff_builder_t(*cluster, file_addr);

            SECTION("source & target are different files") {
                proto::FileInfo pr_target;
                proto::set_name(pr_target, "b.txt");
                proto::set_block_size(pr_target, 5);
                proto::set_version(pr_target, v);

                SECTION("single block target file") {
                    proto::set_size(pr_source, 5);
                    proto::set_size(pr_target, 5);
                    proto::set_modified_s(pr_target, modified);

                    auto source = make_file(pr_source, 1);
                    auto target = make_file(pr_target, 1);

                    auto source_file = folder_peer->get_file_infos().by_name(proto::get_name(pr_source));
                    builder.append_block(*source_file, 0, as_owned_bytes("12345"))
                        .apply(*sup)
                        .finish_file(*source_file)
                        .apply(*sup);

                    auto target_file = folder_peer->get_file_infos().by_name(proto::get_name(pr_target));
                    auto block = source_file->get_blocks()[0];
                    auto file_block = model::file_block_t(block.get(), target_file.get(), 0);
                    builder.clone_block(file_block).apply(*sup).finish_file(*target).apply(*sup);

                    auto path = root_path / std::string(target_file->get_name());
                    REQUIRE(bfs::exists(path));
                    REQUIRE(bfs::file_size(path) == 5);
                    auto data = read_file(path);
                    CHECK(data == "12345");
                    CHECK(to_unix(bfs::last_write_time(path)) == 1641828421);
                }

                SECTION("multi block target file") {
                    proto::set_size(pr_source, 10);
                    proto::set_size(pr_target, 10);
                    auto source = make_file(pr_source, 2);
                    auto target = make_file(pr_target, 2);

                    auto source_file = folder_peer->get_file_infos().by_name(proto::get_name(pr_source));

                    builder.append_block(*source_file, 0, as_owned_bytes("12345"))
                        .append_block(*source_file, 1, as_owned_bytes("67890"))
                        .apply(*sup);

                    auto target_file = folder_peer->get_file_infos().by_name(proto::get_name(pr_target));
                    auto blocks = source_file->get_blocks();
                    auto fb_1 = model::file_block_t(blocks[0].get(), target_file.get(), 0);
                    auto fb_2 = model::file_block_t(blocks[1].get(), target_file.get(), 1);
                    builder.clone_block(fb_1).clone_block(fb_2).apply(*sup).finish_file(*target).apply(*sup);

                    auto filename = std::string(target_file->get_name());
                    auto path = root_path / filename;
                    REQUIRE(bfs::exists(path));
                    REQUIRE(bfs::file_size(path) == 10);
                    auto data = read_file(path);
                    CHECK(data == "1234567890");
                }

                SECTION("source/target different sizes") {
                    proto::set_size(pr_source, 5);
                    proto::set_size(pr_target, 10);
                    auto target = make_file(pr_target, 2);

                    proto::set_sequence(pr_source, folder_peer->get_max_sequence() + 1);
                    auto source = file_info_t::create(sequencer->next_uuid(), pr_source, folder_peer).value();
                    source->assign_block(blocks[1], 0);
                    REQUIRE(folder_peer->add_strict(source));

                    auto source_file = folder_peer->get_file_infos().by_name(proto::get_name(pr_source));
                    auto target_file = folder_peer->get_file_infos().by_name(proto::get_name(pr_target));

                    builder.append_block(*source_file, 0, as_owned_bytes("67890"))
                        .append_block(*target_file, 0, as_owned_bytes("12345"))
                        .apply(*sup);

                    auto blocks = source_file->get_blocks();
                    auto fb = model::file_block_t(blocks[0].get(), target_file.get(), 1);
                    builder.clone_block(fb).apply(*sup).finish_file(*target).apply(*sup);

                    auto filename = std::string(target_file->get_name());
                    auto path = root_path / filename;
                    REQUIRE(bfs::exists(path));
                    REQUIRE(bfs::file_size(path) == 10);
                    auto data = read_file(path);
                    CHECK(data == "1234567890");
                }
            }

            SECTION("source & target are is the same file") {
                proto::set_sequence(pr_source, folder_peer->get_max_sequence() + 1);
                proto::set_size(pr_source, 10);

                auto source = file_info_t::create(sequencer->next_uuid(), pr_source, folder_peer).value();
                source->assign_block(blocks[0], 0);
                source->assign_block(blocks[0], 1);
                REQUIRE(folder_peer->add_strict(source));

                auto source_file = folder_peer->get_file_infos().by_name(proto::get_name(pr_source));
                auto target_file = source_file;

                builder.append_block(*source_file, 0, as_owned_bytes("12345")).apply(*sup);

                auto block = source_file->get_blocks()[0];
                auto file_block = model::file_block_t(block.get(), target_file.get(), 1);
                builder.clone_block(file_block).apply(*sup).finish_file(*source).apply(*sup);

                auto path = root_path / std::string(target_file->get_name());
                REQUIRE(bfs::exists(path));
                REQUIRE(bfs::file_size(path) == 10);
                auto data = read_file(path);
                CHECK(data == "1234512345");
                CHECK(to_unix(bfs::last_write_time(path)) == 1641828421);
            }
        }
    };
    F().run();
}

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

            auto hash_1 = utils::sha256_digest(as_bytes("12345")).value();
            auto bi = proto::BlockInfo();
            proto::set_size(bi, 5);
            proto::set_hash(bi, hash_1);
            auto b = block_info_t::create(bi).value();

            auto hash_2 = utils::sha256_digest(as_bytes("67890")).value();
            auto bi2 = proto::BlockInfo();
            proto::set_size(bi2, 5);
            proto::set_hash(bi2, hash_2);
            auto b2 = block_info_t::create(bi2).value();

            cluster->get_blocks().put(b);
            cluster->get_blocks().put(b2);
            bfs::path target = root_path / "a.txt";

            std::int64_t modified = 1641828421;
            proto::FileInfo pr_source;

            proto::set_name(pr_source, "a.txt");
            proto::set_modified_s(pr_source, modified);
            proto::set_block_size(pr_source, 5);
            proto::set_size(pr_source, 10);
            proto::set_sequence(pr_source, folder_my->get_max_sequence() + 1);

            auto &v = proto::get_version(pr_source);
            proto::add_counters(v, proto::Counter(peer_device->device_id().get_uint(), 1));

            proto::add_blocks(pr_source, bi);
            proto::add_blocks(pr_source, bi2);

            auto file = file_info_t::create(sequencer->next_uuid(), pr_source, folder_my).value();
            file->assign_block(b, 0);
            file->assign_block(b2, 1);
            folder_my->add_relaxed(file);

            auto req = proto::Request();
            proto::set_folder(req, folder->get_id());
            proto::set_name(req, "a.txt");
            proto::set_size(req, 5);

            auto msg =
                r::make_message<fs::payload::block_request_t>(file_actor->get_address(), req, sup->get_address());

            SECTION("error, no file") {
                sup->put(msg);
                sup->do_process();
                REQUIRE(block_reply);
                REQUIRE(block_reply->payload.ec);
                REQUIRE(block_reply->payload.data.empty());
            }

            SECTION("error, oversized request") {
                write_file(target, "1234");
                sup->put(msg);
                sup->do_process();
                REQUIRE(block_reply);
                REQUIRE(block_reply->payload.ec);
                REQUIRE(block_reply->payload.data.empty());
            }

            SECTION("successful file reading") {
                write_file(target, "1234567890");
                sup->put(msg);
                sup->do_process();
                REQUIRE(block_reply);
                REQUIRE(!block_reply->payload.ec);
                REQUIRE(block_reply->payload.data == as_bytes("12345"));

                proto::set_offset(req, 5);
                auto msg =
                    r::make_message<fs::payload::block_request_t>(file_actor->get_address(), req, sup->get_address());
                sup->put(msg);
                sup->do_process();
                REQUIRE(block_reply);
                REQUIRE(!block_reply->payload.ec);
                REQUIRE(block_reply->payload.data == as_bytes("67890"));
            }
        }
    };
    F().run();
}

void test_conflicts() {
    struct F : fixture_t {
        void main() noexcept override {
            auto builder = diff_builder_t(*cluster, file_addr);
            auto &blocks_map = cluster->get_blocks();

            proto::FileInfo pr_fi;
            std::int64_t modified = 1641828421;
            proto::set_name(pr_fi, "q.txt");
            proto::set_modified_s(pr_fi, modified);
            proto::set_sequence(pr_fi, folder_peer->get_max_sequence() + 1);

            SECTION("non-empty (via file_finish)") {
                proto::set_block_size(pr_fi, 5);
                proto::set_size(pr_fi, 5);

                auto peer_block = []() {
                    auto block = proto::BlockInfo();
                    auto hash = utils::sha256_digest(as_bytes("12345")).value();
                    proto::set_hash(block, hash);
                    proto::set_size(block, 5);
                    auto bi = block_info_t::create(block).value();
                    return bi;
                }();
                blocks_map.put(peer_block);

                auto my_block = []() {
                    auto block = proto::BlockInfo();
                    auto hash = utils::sha256_digest(as_bytes("67890")).value();
                    proto::set_hash(block, hash);
                    proto::set_size(block, 5);
                    auto bi = block_info_t::create(block).value();
                    return bi;
                }();
                blocks_map.put(my_block);

                SECTION("remote win") {
                    auto peer_file = [&]() {
                        auto &v = proto::get_version(pr_fi);
                        auto &c = proto::add_counters(v);
                        proto::set_id(c, peer_device->device_id().get_uint());
                        proto::set_value(c, 10);
                        proto::add_blocks(pr_fi, peer_block->as_bep(0));
                        proto::set_modified_s(pr_fi, 1734690000);

                        auto file = file_info_t::create(sequencer->next_uuid(), pr_fi, folder_peer).value();
                        REQUIRE(folder_peer->add_strict(file));
                        auto vv = &proto::get_version(pr_fi);
                        proto::clear_counters(*vv);
                        proto::clear_blocks(pr_fi);
                        file->assign_block(peer_block, 0);
                        return file;
                    }();

                    auto my_file = [&]() {
                        proto:
                            auto &v = proto::get_version(pr_fi);
                            proto::add_counters(v, proto::Counter(my_device->device_id().get_uint(), 5));
                            proto::set_modified_s(pr_fi, 1734600000);
                            proto::add_blocks(pr_fi, my_block->as_bep(0));
                            auto file = file_info_t::create(sequencer->next_uuid(), pr_fi, folder_my).value();
                            REQUIRE(folder_my->add_strict(file));
                            file->mark_local();

                            file->assign_block(my_block, 0);
                            return file;
                    }();

                    bfs::path kept_file = root_path / proto::get_name(pr_fi);
                    write_file(kept_file, "12345");

                    REQUIRE(model::resolve(*peer_file) == advance_action_t::resolve_remote_win);
                    builder.append_block(*peer_file, 0, as_owned_bytes("67890"))
                        .apply(*sup)
                        .finish_file(*peer_file)
                        .apply(*sup);

                    auto conflict_file = root_path / my_file->make_conflicting_name();
                    CHECK(read_file(kept_file) == "67890");
                    CHECK(bfs::exists(conflict_file));
                    CHECK(read_file(conflict_file) == "12345");
                }
            }
            SECTION("remote win emtpy (file vs directory)") {
                auto peer_file = [&]() {
                    auto &v = proto::get_version(pr_fi);
                    proto::add_counters(v, proto::Counter(peer_device->device_id().get_uint(), 10));
                    proto::set_modified_s(pr_fi, 1734690000);
                    auto file = file_info_t::create(sequencer->next_uuid(), pr_fi, folder_peer).value();
                    REQUIRE(folder_peer->add_strict(file));
                    auto vv = &proto::get_version(pr_fi);
                    proto::clear_counters(*vv);
                    return file;
                }();

                auto my_file = [&]() {
                    auto &v = proto::get_version(pr_fi);
                    proto::add_counters(v, proto::Counter(my_device->device_id().get_uint(), 5));
                    proto::set_modified_s(pr_fi, 1734600000);
                    auto file = file_info_t::create(sequencer->next_uuid(), pr_fi, folder_my).value();
                    REQUIRE(folder_my->add_strict(file));
                    file->mark_local();
                    return file;
                }();

                bfs::path kept_file = root_path / proto::get_name(pr_fi);
                bfs::create_directories(kept_file);

                builder.advance(*peer_file).apply(*sup);

                auto conflict_file = root_path / my_file->make_conflicting_name();
                CHECK(bfs::exists(conflict_file));
                CHECK(bfs::exists(kept_file));
                CHECK(bfs::is_directory(conflict_file));
                CHECK(bfs::is_regular_file(kept_file));
            }
        }
    };
    F().run();
}

int _init() {
    REGISTER_TEST_CASE(test_remote_copy, "test_remote_copy", "[fs]");
    REGISTER_TEST_CASE(test_append_block, "test_append_block", "[fs]");
    REGISTER_TEST_CASE(test_clone_block, "test_clone_block", "[fs]");
    REGISTER_TEST_CASE(test_requesting_block, "test_requesting_block", "[fs]");
    REGISTER_TEST_CASE(test_conflicts, "test_conflicts", "[fs]");
    return 1;
}

static int v = _init();


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

#include "test-utils.h"
#include "diff-builder.h"
#include "access.h"
#include "model/cluster.h"
#include "model/diff/contact/update_contact.h"
#include "utils/tls.h"
#include "utils/error_code.h"
#include "net/global_discovery_actor.h"
#include "net/names.h"
#include "net/messages.h"
#include "test_supervisor.h"

#include <nlohmann/json.hpp>

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

namespace http = boost::beast::http;
using json = nlohmann::json;

namespace {

static auto ssl_pair = utils::generate_pair("sample").value();

struct dummy_http_actor_t : r::actor_base_t {
    using response_t = r::intrusive_ptr_t<net::payload::http_response_t>;
    using queue_t = std::list<response_t>;

    using r::actor_base_t::actor_base_t;

    void configure(r::plugin::plugin_base_t &plugin) noexcept override {
        r::actor_base_t::configure(plugin);
        plugin.with_casted<r::plugin::address_maker_plugin_t>([&](auto &p) { p.set_identity("dummy-http", true); });
        plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
            p.subscribe_actor(&dummy_http_actor_t::on_request);
            p.subscribe_actor(&dummy_http_actor_t::on_close_connection);
        });
        plugin.with_casted<r::plugin::registry_plugin_t>(
            [&](auto &p) { p.register_name(names::http11_gda, get_address()); });
    }

    void on_request(net::message::http_request_t &req) noexcept {
        if (!responses.empty()) {
            auto &res = *responses.front();
            reply_to(req, std::move(res.response), res.bytes, std::move(res.local_addr));
            connected = true;
            responses.pop_front();
        } else {
            auto ec = utils::make_error_code(utils::error_code_t::timed_out);
            reply_with_error(req, make_error(ec));
        }
    }

    void on_close_connection(net::message::http_close_connection_t &) noexcept { closed = true; }

    queue_t responses;
    bool connected = false;
    bool closed = false;
};

struct fixture_t {
    using http_actor_ptr_t = r::intrusive_ptr_t<dummy_http_actor_t>;
    using announce_msg_t = net::message::announce_notification_t;
    using announce_ptr_t = r::intrusive_ptr_t<announce_msg_t>;

    fixture_t() noexcept { test::init_logging(); }

    virtual void run() noexcept {
        auto peer_id =
            device_id_t::from_string("VUV42CZ-IQD5A37-RPEBPM4-VVQK6E4-6WSKC7B-PVJQHHD-4PZD44V-ENC6WAZ").value();
        peer_device = device_t::create(peer_id, "peer-device").value();

        auto my_id =
            device_id_t::from_string("KHQNO2S-5QSILRK-YX4JZZ4-7L77APM-QNVGZJT-EKU7IFI-PNEPBMY-4MXFMQD").value();
        auto my_device = device_t::create(my_id, "my-device").value();
        cluster = new cluster_t(my_device, 1);

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

        r::system_context_t ctx;
        sup = ctx.create_supervisor<supervisor_t>().timeout(timeout).create_registry().finish();
        sup->cluster = cluster;
        sup->configure_callback = [&](r::plugin::plugin_base_t &plugin) {
            plugin.template with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
                p.subscribe_actor(r::lambda<announce_msg_t>([&](announce_msg_t &msg) { announce = &msg; }));
            });
        };
        sup->start();
        http_actor = sup->create_actor<dummy_http_actor_t>().timeout(timeout).finish();
        sup->do_process();

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

        auto global_device_id =
            model::device_id_t::from_string("LYXKCHX-VI3NYZR-ALCJBHF-WMZYSPK-QG6QJA3-MPFYMSO-U56GTUK-NA2MIAW");

        gda = sup->create_actor<global_discovery_actor_t>()
                  .cluster(cluster)
                  .ssl_pair(&ssl_pair)
                  .announce_url(utils::parse("https://discovery.syncthing.net/"))
                  .device_id(std::move(global_device_id.value()))
                  .rx_buff_size(32768ul)
                  .io_timeout(5ul)
                  .timeout(timeout)
                  .finish();
        bool started = preprocess();
        sup->do_process();

        if (started) {
            CHECK(static_cast<r::actor_base_t *>(gda.get())->access<to::state>() == r::state_t::OPERATIONAL);
            target_addr = gda->get_address();
            main();
        }

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

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

    virtual bool preprocess() noexcept { return true; }

    virtual void main() noexcept {}

    http_actor_ptr_t http_actor;
    r::address_ptr_t target_addr;
    r::pt::time_duration timeout = r::pt::millisec{10};
    cluster_ptr_t cluster;
    device_ptr_t peer_device;
    r::intrusive_ptr_t<supervisor_t> sup;
    r::intrusive_ptr_t<net::global_discovery_actor_t> gda;
    announce_ptr_t announce;
    r::system_context_t ctx;
};

} // namespace

void test_successful_announcement() {
    struct F : fixture_t {
        bool preprocess() noexcept override {
            auto uri = utils::parse("tcp://127.0.0.1");
            cluster->get_device()->assign_uris({uri});

            SECTION("successful (and empty) announce response") {
                http::response<http::string_body> res;
                res.result(204);
                res.set("Reannounce-After", "123");
                http_actor->responses.push_back(new net::payload::http_response_t(std::move(res), 0));
                sup->do_process();
                CHECK(http_actor->connected);
                CHECK(announce);
            }

            return true;
        }
    };
    F().run();
}

void test_failed_announcement() {
    struct F : fixture_t {
        bool preprocess() noexcept override {
            auto uri = utils::parse("tcp://127.0.0.1");
            cluster->get_device()->assign_uris({uri});

            SECTION("successful (and empty) announce response") {
                http::response<http::string_body> res;
                res.result(204);
                http_actor->responses.push_back(new net::payload::http_response_t(std::move(res), 0));
                sup->do_process();
                CHECK(http_actor->connected);
                CHECK(!announce);
                CHECK(static_cast<r::actor_base_t *>(gda.get())->access<to::state>() == r::state_t::SHUT_DOWN);
            }

            return false;
        }
    };
    F().run();
}

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

            auto builder = diff_builder_t(*cluster);
            http::response<http::string_body> res;

            auto j = json::object();
            j["addresses"] = json::array({"tcp://127.0.0.2"});
            j["seen"] = "2020-10-13T18:41:37.02287354Z";

            SECTION("successful case") {
                res.body() = j.dump();

                http_actor->responses.push_back(new net::payload::http_response_t(std::move(res), 0));
                builder.update_state(*peer_device, {}, model::device_state_t::discovering).apply(*sup);

                REQUIRE(peer_device->get_uris().size() == 1);
                CHECK(peer_device->get_uris()[0]->buffer() == "tcp://127.0.0.2");
                REQUIRE(peer_device->get_state() == model::device_state_t::discovering);

                // 2nd attempt
                peer_device->assign_uris({});
                res = {};
                res.body() = j.dump();
                http_actor->responses.push_back(new net::payload::http_response_t(std::move(res), 0));
                builder.update_state(*peer_device, {}, model::device_state_t::discovering).apply(*sup);

                REQUIRE(peer_device->get_uris().size() == 1);
                CHECK(peer_device->get_uris()[0]->buffer() == "tcp://127.0.0.2");
                REQUIRE(peer_device->get_state() == model::device_state_t::discovering);

                // 3nd attempt (empty urls)
                j["addresses"] = json::array();
                peer_device->assign_uris({});
                res = {};
                res.body() = j.dump();
                builder.update_state(*peer_device, {}, model::device_state_t::discovering).apply(*sup);
                REQUIRE(peer_device->get_uris().size() == 0);
                REQUIRE(peer_device->get_state() == model::device_state_t::offline);
            }

            SECTION("gargbage in response") {
                http_actor->responses.push_back(new net::payload::http_response_t(std::move(res), 0));
                builder.update_state(*peer_device, {}, model::device_state_t::discovering).apply(*sup);

                REQUIRE(peer_device->get_uris().size() == 0);
                CHECK(static_cast<r::actor_base_t *>(gda.get())->access<to::state>() == r::state_t::OPERATIONAL);
                REQUIRE(peer_device->get_state() == model::device_state_t::offline);
            }
        }
    };
    F().run();
}

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

            auto diff = model::diff::cluster_diff_ptr_t{};
            auto url_1 = utils::parse("tcp://127.0.0.3:22000");
            auto &self = cluster->get_device()->device_id();
            diff = new model::diff::contact::update_contact_t(*cluster, self, {url_1});
            sup->send<model::payload::model_update_t>(sup->get_address(), diff);

            http::response<http::string_body> res;
            res.result(204);
            res.set("Reannounce-After", "123");
            http_actor->responses.push_back(new net::payload::http_response_t(std::move(res), 0));

            sup->do_process();
            CHECK(http_actor->connected);
            CHECK(announce);
        }
    };
    F().run();
}

int _init() {
    REGISTER_TEST_CASE(test_successful_announcement, "test_successful_announcement", "[net]");
    REGISTER_TEST_CASE(test_failed_announcement, "test_failed_announcement", "[net]");
    REGISTER_TEST_CASE(test_peer_discovery, "test_peer_discovery", "[net]");
    REGISTER_TEST_CASE(test_late_announcement, "test_late_announcement", "[net]");
    return 1;
}

static int v = _init();


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

#include "test-utils.h"
#include "access.h"
#include "model/cluster.h"
#include "model/diff/contact/dial_request.h"
#include "diff-builder.h"

#include "net/dialer_actor.h"
#include "access.h"
#include "test_supervisor.h"

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

using state_t = model::device_state_t;

namespace {

struct fixture_t : private model::diff::cluster_visitor_t {
    using msg_t = model::message::model_update_t;
    using msg_ptr_t = r::intrusive_ptr_t<msg_t>;
    using messages_t = std::vector<msg_ptr_t>;

    fixture_t(bool start_dialer_) noexcept : start_dialer{start_dialer_} { test::init_logging(); }

    virtual void run() noexcept {
        auto peer_id =
            device_id_t::from_string("VUV42CZ-IQD5A37-RPEBPM4-VVQK6E4-6WSKC7B-PVJQHHD-4PZD44V-ENC6WAZ").value();
        peer_device = device_t::create(peer_id, "peer-device").value();

        auto my_id =
            device_id_t::from_string("KHQNO2S-5QSILRK-YX4JZZ4-7L77APM-QNVGZJT-EKU7IFI-PNEPBMY-4MXFMQD").value();
        auto my_device = device_t::create(my_id, "my-device").value();
        cluster = new cluster_t(my_device, 1);

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

        r::system_context_t ctx;
        sup = ctx.create_supervisor<supervisor_t>().timeout(timeout).create_registry().finish();
        sup->cluster = cluster;
        sup->configure_callback = [&](r::plugin::plugin_base_t &plugin) {
            plugin.template with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
                p.subscribe_actor(r::lambda<msg_t>([&](msg_t &msg) {
                    std::ignore = msg.payload.diff->apply(*cluster, get_apply_controller());
                    messages.emplace_back(&msg);
                }));
            });
        };

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

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

        auto global_device_id =
            model::device_id_t::from_string("LYXKCHX-VI3NYZR-ALCJBHF-WMZYSPK-QG6QJA3-MPFYMSO-U56GTUK-NA2MIAW");

        auto cfg = config::dialer_config_t{true, 500, 1};
        auto dialer = sup->create_actor<dialer_actor_t>().cluster(cluster).dialer_config(cfg).timeout(timeout).finish();
        target_addr = dialer->get_address();
        if (start_dialer) {
            sup->do_process();
            CHECK(static_cast<r::actor_base_t *>(dialer.get())->access<to::state>() == r::state_t::OPERATIONAL);
        }
        main();

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

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

    virtual void main() noexcept {}

    bool start_dialer;
    r::address_ptr_t target_addr;
    r::pt::time_duration timeout = r::pt::millisec{10};
    cluster_ptr_t cluster;
    device_ptr_t peer_device;
    r::intrusive_ptr_t<supervisor_t> sup;
    r::system_context_t ctx;
    messages_t messages;
};

} // namespace

void test_dialer() {
    struct F : fixture_t {
        using fixture_t::fixture_t;
        void main() noexcept override {
            auto builder = diff_builder_t(*cluster);

            REQUIRE(messages.empty());
            REQUIRE(peer_device->get_state() == state_t::offline);

            sup->send<net::payload::announce_notification_t>(sup->get_address());
            sup->do_process();

            SECTION("peer is not online => discover it on timeout") {
                REQUIRE(messages.size() == 1);
                REQUIRE(peer_device->get_state() == state_t::discovering);
            }

            SECTION("peer online & offline") {
                messages.clear();
                auto connection_id = std::string("tcp://127.0.0.1:1234");
                builder.update_state(*peer_device, {}, model::device_state_t::online, connection_id).apply(*sup);
                CHECK(messages.size() == 1);

                builder.update_state(*peer_device, {}, model::device_state_t::offline, connection_id).apply(*sup);
                CHECK(messages.size() == 2);
                CHECK(sup->timers.size() == 1);

                sup->do_invoke_timer((*sup->timers.begin())->request_id);
                sup->do_process();
                CHECK(messages.size() == 3);
                CHECK(peer_device->get_state() == state_t::discovering);
                CHECK(sup->timers.size() == 0);

                auto uri = utils::parse("tcp://127.0.0.1");
                builder.update_contact(peer_device->device_id(), {uri}).apply(*sup);
                REQUIRE(messages.size() == 5);
                CHECK(peer_device->get_state() == state_t::discovering);
                CHECK(sup->timers.size() == 0);
                auto diff = messages.back()->payload.diff;
                REQUIRE(dynamic_cast<diff::contact::dial_request_t *>(diff.get()));

                builder.update_state(*peer_device, {}, model::device_state_t::offline).apply(*sup);
                CHECK(sup->timers.size() == 1);
                sup->do_invoke_timer((*sup->timers.begin())->request_id);
                sup->do_process();
                REQUIRE(messages.size() == 7);
                CHECK(peer_device->get_state() == state_t::offline);
                CHECK(sup->timers.size() == 0);
                diff = messages.back()->payload.diff;
                REQUIRE(dynamic_cast<diff::contact::dial_request_t *>(diff.get()));
            }

            SECTION("remove peer") {
                SECTION("start discover") {
                    builder.update_state(*peer_device, {}, model::device_state_t::offline).apply(*sup);
                    CHECK(messages.size() == 2);
                    CHECK(sup->timers.size() == 1);
                }

                builder.remove_peer(*peer_device).apply(*sup);
                CHECK(sup->timers.size() == 0);
            }
        }
    };
    F(true).run();
}

void test_static_address() {
    struct F : fixture_t {
        using fixture_t::fixture_t;
        void main() noexcept override {
            auto builder = diff_builder_t(*cluster);

            REQUIRE(messages.empty());
            REQUIRE(peer_device->get_state() == state_t::offline);
            auto uri = utils::parse("tcp://127.0.0.1");
            peer_device->set_static_uris({uri});

            sup->do_process();
            REQUIRE(peer_device->get_state() == state_t::offline);
            REQUIRE(messages.size() == 2);
            auto diff = messages.back()->payload.diff;
            REQUIRE(dynamic_cast<diff::contact::dial_request_t *>(diff.get()));

            builder.update_state(*peer_device, {}, model::device_state_t::offline).apply(*sup);
            CHECK(sup->timers.size() == 1);

            SECTION("remove") {
                builder.remove_peer(*peer_device).apply(*sup);
                CHECK(sup->timers.size() == 0);
                REQUIRE(messages.size() == 4);
            }

            SECTION("invoke") {
                sup->do_invoke_timer((*sup->timers.begin())->request_id);
                sup->do_process();
                REQUIRE(messages.size() == 4);

                auto diff = messages.back()->payload.diff;
                REQUIRE(dynamic_cast<diff::contact::dial_request_t *>(diff.get()));
            }
        }
    };
    F(false).run();
}

void test_peer_removal() {
    struct F : fixture_t {
        using fixture_t::fixture_t;
        void main() noexcept override {
            REQUIRE(messages.empty());
            REQUIRE(peer_device->get_state() == state_t::offline);

            SECTION("with announce") {
                sup->send<net::payload::announce_notification_t>(sup->get_address());
                sup->do_process();

                diff_builder_t(*cluster).remove_peer(*peer_device).apply(*sup);
                REQUIRE(sup->timers.size() == 0);
            }

            SECTION("without announce") {
                diff_builder_t(*cluster).remove_peer(*peer_device).apply(*sup);
                REQUIRE(sup->timers.size() == 0);
            }
        }
    };
    F(true).run();
}

int _init() {
    REGISTER_TEST_CASE(test_dialer, "test_dialer", "[net]");
    REGISTER_TEST_CASE(test_static_address, "test_static_address", "[net]");
    REGISTER_TEST_CASE(test_peer_removal, "test_peer_removal", "[net]");
    return 1;
}

static int v = _init();