#include <test/jtx/Env.h>
#include <test/jtx/envconfig.h>

#include <xrpld/app/main/Application.h>
#include <xrpld/overlay/Message.h>
#include <xrpld/overlay/Peer.h>
#include <xrpld/overlay/ReduceRelayCommon.h>
#include <xrpld/overlay/Slot.h>
#include <xrpld/overlay/Squelch.h>
#include <xrpld/overlay/detail/Handshake.h>

#include <xrpl/basics/base_uint.h>
#include <xrpl/basics/random.h>
#include <xrpl/beast/net/IPAddress.h>
#include <xrpl/beast/net/IPEndpoint.h>
#include <xrpl/beast/unit_test/suite.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/json/json_value.h>
#include <xrpl/protocol/KeyType.h>
#include <xrpl/protocol/PublicKey.h>
#include <xrpl/protocol/SecretKey.h>

#include <boost/asio/ip/address.hpp>

#include <xrpl.pb.h>

#include <algorithm>
#include <cassert>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <iostream>
#include <iterator>
#include <memory>
#include <numeric>
#include <optional>
#include <random>
#include <ratio>
#include <set>
#include <sstream>
#include <stdexcept>
#include <string>
#include <tuple>
#include <unordered_map>
#include <utility>
#include <vector>

namespace xrpl::test {

using namespace std::chrono;

class Link;

using MessageSPtr = std::shared_ptr<Message>;
using LinkSPtr = std::shared_ptr<Link>;
using PeerSPtr = std::shared_ptr<Peer>;
using PeerWPtr = std::weak_ptr<Peer>;
using SquelchCB = std::function<void(PublicKey const&, PeerWPtr const&, std::uint32_t)>;
using UnsquelchCB = std::function<void(PublicKey const&, PeerWPtr const&)>;
using LinkIterCB = std::function<void(Link&, MessageSPtr)>;

static constexpr std::uint32_t kMaxPeers = 10;
static constexpr std::uint32_t kMaxValidators = 10;
static constexpr std::uint32_t kMaxMessages = 200000;

/**
 * Simulate two entities - peer directly connected to the server
 * (via squelch in PeerSim) and PeerImp (via Overlay)
 */
class PeerPartial : public Peer
{
public:
    PeerPartial() : nodePublicKey(derivePublicKey(KeyType::Ed25519, randomSecretKey()))
    {
    }

    PublicKey nodePublicKey;
    ~PeerPartial() override = default;
    virtual void
    onMessage(MessageSPtr const& m, SquelchCB f) = 0;
    virtual void
    onMessage(protocol::TMSquelch const& squelch) = 0;
    void
    send(protocol::TMSquelch const& squelch)
    {
        onMessage(squelch);
    }

    // dummy implementation
    void
    send(std::shared_ptr<Message> const& m) override
    {
    }
    [[nodiscard]] beast::IP::Endpoint
    getRemoteAddress() const override
    {
        return {};
    }
    void
    charge(Resource::Charge const& fee, std::string const& context = {}) override
    {
    }
    [[nodiscard]] bool
    cluster() const override
    {
        return false;
    }
    [[nodiscard]] bool
    isHighLatency() const override
    {
        return false;
    }
    [[nodiscard]] int
    getScore(bool) const override
    {
        return 0;
    }
    [[nodiscard]] PublicKey const&
    getNodePublic() const override
    {
        return nodePublicKey;
    }
    json::Value
    json() override
    {
        return {};
    }
    [[nodiscard]] bool
    supportsFeature(ProtocolFeature f) const override
    {
        return false;
    }
    [[nodiscard]] std::optional<std::size_t>
    publisherListSequence(PublicKey const&) const override
    {
        return {};
    }
    void
    setPublisherListSequence(PublicKey const&, std::size_t const) override
    {
    }
    [[nodiscard]] uint256
    getClosedLedgerHash() const override
    {
        static uint256 const kHash{};
        return kHash;
    }
    [[nodiscard]] bool
    hasLedger(uint256 const& hash, std::uint32_t seq) const override
    {
        return false;
    }
    void
    ledgerRange(std::uint32_t& minSeq, std::uint32_t& maxSeq) const override
    {
    }
    [[nodiscard]] bool
    hasTxSet(uint256 const& hash) const override
    {
        return false;
    }
    void
    cycleStatus() override
    {
    }
    bool
    hasRange(std::uint32_t uMin, std::uint32_t uMax) override
    {
        return false;
    }
    [[nodiscard]] bool
    compressionEnabled() const override
    {
        return false;
    }
    [[nodiscard]] bool
    txReduceRelayEnabled() const override
    {
        return false;
    }
    void
    sendTxQueue() override
    {
    }
    void
    addTxQueue(uint256 const&) override
    {
    }
    void
    removeTxQueue(uint256 const&) override
    {
    }
};

/**
 * Manually advanced clock.
 */
class ManualClock
{
public:
    using rep = uint64_t;
    using period = std::milli;
    using duration = std::chrono::duration<std::uint32_t, period>;
    using time_point = std::chrono::time_point<ManualClock>;
    inline static bool const is_steady = false;  // NOLINT(readability-identifier-naming)

    static void
    advance(duration d) noexcept
    {
        kNow += d;
    }

    static void
    randAdvance(milliseconds min, milliseconds max)
    {
        kNow += randDuration(min, max);
    }

    static void
    reset() noexcept
    {
        kNow = time_point(seconds(0));
    }

    static time_point
    now() noexcept
    {
        return kNow;
    }

    static duration
    randDuration(milliseconds min, milliseconds max)
    {
        return duration(milliseconds(randInt(min.count(), max.count())));
    }

    explicit ManualClock() = default;

private:
    inline static time_point kNow = time_point(seconds(0));
};

/**
 * Simulate server's OverlayImpl
 */
class Overlay
{
public:
    Overlay() = default;
    virtual ~Overlay() = default;

    virtual void
    updateSlotAndSquelch(
        uint256 const& key,
        PublicKey const& validator,
        Peer::id_t id,
        SquelchCB f,
        protocol::MessageType type = protocol::mtVALIDATION) = 0;

    virtual void deleteIdlePeers(UnsquelchCB) = 0;

    virtual void deletePeer(Peer::id_t, UnsquelchCB) = 0;
};

class Validator;

/**
 * Simulate link from a validator to a peer directly connected
 * to the server.
 */
class Link
{
    using Latency = std::pair<milliseconds, milliseconds>;

public:
    Link(Validator& validator, PeerSPtr peer, Latency latency = {milliseconds(5), milliseconds(15)})
        : validator_(validator), peer_(peer), latency_(std::move(latency))
    {
        auto sp = peer_.lock();
        assert(sp);
    }
    ~Link() = default;
    void
    send(MessageSPtr const& m, SquelchCB f)
    {
        if (!up_)
            return;
        auto sp = peer_.lock();
        assert(sp);
        auto peer = std::dynamic_pointer_cast<PeerPartial>(sp);
        peer->onMessage(m, f);
    }
    Validator&
    validator()
    {
        return validator_;
    }
    void
    up(bool linkUp)
    {
        up_ = linkUp;
    }
    Peer::id_t
    peerId()
    {
        auto p = peer_.lock();
        assert(p);
        return p->id();
    }
    PeerSPtr
    getPeer()
    {
        auto p = peer_.lock();
        assert(p);
        return p;
    }

private:
    Validator& validator_;
    PeerWPtr peer_;
    Latency latency_;
    bool up_{true};
};

/**
 * Simulate Validator
 */
class Validator
{
    using Links = std::unordered_map<Peer::id_t, LinkSPtr>;

public:
    Validator() : pkey_(std::get<0>(randomKeyPair(KeyType::Ed25519))), id_(sid++)
    {
        protocol::TMValidation v;
        v.set_validation("validation");
        message_ = std::make_shared<Message>(v, protocol::mtVALIDATION, pkey_);
    }
    Validator(Validator const&) = default;
    Validator(Validator&&) = default;
    Validator&
    operator=(Validator const&) = default;
    Validator&
    operator=(Validator&&) = default;
    ~Validator()
    {
        clear();
    }

    void
    clear()
    {
        links_.clear();
    }

    static void
    resetId()
    {
        sid = 0;
    }

    PublicKey const&
    key()
    {
        return pkey_;
    }

    operator PublicKey() const
    {
        return pkey_;
    }

    void
    addPeer(PeerSPtr peer)
    {
        links_.emplace(std::make_pair(peer->id(), std::make_shared<Link>(*this, peer)));
    }

    void
    deletePeer(Peer::id_t id)
    {
        links_.erase(id);
    }

    void
    forLinks(std::vector<Peer::id_t> peers, LinkIterCB f)
    {
        for (auto id : peers)
        {
            assert(links_.contains(id));
            f(*links_[id], message_);
        }
    }

    void
    forLinks(LinkIterCB f, bool simulateSlow = false)
    {
        std::vector<LinkSPtr> v;
        std::ranges::transform(links_, std::back_inserter(v), [](auto& kv) { return kv.second; });
        std::random_device d;
        std::mt19937 g(d());
        std::shuffle(v.begin(), v.end(), g);

        for (auto& link : v)
        {
            f(*link, message_);
        }
    }

    /**
     * Send to specific peers
     */
    void
    send(std::vector<Peer::id_t> peers, SquelchCB f)
    {
        forLinks(peers, [&](Link& link, MessageSPtr m) { link.send(m, f); });
    }

    /**
     * Send to all peers
     */
    void
    send(SquelchCB f)
    {
        forLinks([&](Link& link, MessageSPtr m) { link.send(m, f); });
    }

    MessageSPtr
    message()
    {
        return message_;
    }

    [[nodiscard]] std::uint16_t
    id() const
    {
        return id_;
    }

    void
    linkUp(Peer::id_t id)
    {
        auto it = links_.find(id);
        assert(it != links_.end());
        it->second->up(true);
    }

    void
    linkDown(Peer::id_t id)
    {
        auto it = links_.find(id);
        assert(it != links_.end());
        it->second->up(false);
    }

private:
    Links links_;
    PublicKey pkey_;
    MessageSPtr message_ = nullptr;
    inline static std::uint16_t sid = 0;
    std::uint16_t id_ = 0;
};

class PeerSim : public PeerPartial, public std::enable_shared_from_this<PeerSim>
{
public:
    using id_t = Peer::id_t;
    PeerSim(Overlay& overlay, beast::Journal journal) : overlay_(overlay), squelch_(journal)
    {
    }

    ~PeerSim() override = default;

    id_t
    id() const override
    {
        return id_;
    }

    std::string const&
    fingerprint() const override
    {
        return fingerprint_;
    }

    static void
    resetId()
    {
        sid = 0;
    }

    /**
     * Local Peer (PeerImp)
     */
    void
    onMessage(MessageSPtr const& m, SquelchCB f) override
    {
        auto validator = m->getValidatorKey();
        assert(validator);
        if (!squelch_.expireSquelch(*validator))  // NOLINT(bugprone-unchecked-optional-access)
            return;

        overlay_.updateSlotAndSquelch(
            {}, *validator, id(), f);  // NOLINT(bugprone-unchecked-optional-access)
    }

    /**
     * Remote Peer (Directly connected Peer)
     */
    void
    onMessage(protocol::TMSquelch const& squelch) override
    {
        auto validator = squelch.validatorpubkey();
        PublicKey const key(Slice(validator.data(), validator.size()));
        if (squelch.squelch())
        {
            squelch_.addSquelch(key, std::chrono::seconds{squelch.squelchduration()});
        }
        else
        {
            squelch_.removeSquelch(key);
        }
    }

private:
    inline static id_t sid = 0;
    std::string fingerprint_;
    id_t id_{sid++};
    Overlay& overlay_;
    reduce_relay::Squelch<ManualClock> squelch_;
};

class OverlaySim : public Overlay, public reduce_relay::SquelchHandler
{
    using Peers = std::unordered_map<Peer::id_t, PeerSPtr>;

public:
    using id_t = Peer::id_t;
    using clock_type = ManualClock;
    OverlaySim(Application& app) : slots_(app, *this, app.config()), registry_(app)
    {
    }

    ~OverlaySim() override = default;

    void
    clear()
    {
        peers_.clear();
        ManualClock::advance(hours(1));
        slots_.deleteIdlePeers();
    }

    std::uint16_t
    inState(PublicKey const& validator, reduce_relay::PeerState state)
    {
        auto res = slots_.inState(validator, state);
        return res ? *res : 0;
    }

    void
    updateSlotAndSquelch(
        uint256 const& key,
        PublicKey const& validator,
        Peer::id_t id,
        SquelchCB f,
        protocol::MessageType type = protocol::mtVALIDATION) override
    {
        squelch_ = f;
        slots_.updateSlotAndSquelch(key, validator, id, type);
    }

    void
    deletePeer(id_t id, UnsquelchCB f) override
    {
        unsquelch_ = f;
        slots_.deletePeer(id, true);
    }

    void
    deleteIdlePeers(UnsquelchCB f) override
    {
        unsquelch_ = f;
        slots_.deleteIdlePeers();
    }

    PeerSPtr
    addPeer(bool useCache = true)
    {
        PeerSPtr peer{};
        Peer::id_t id = 0;
        if (peersCache_.empty() || !useCache)
        {
            peer = std::make_shared<PeerSim>(*this, registry_.getJournal("Squelch"));
            id = peer->id();
        }
        else
        {
            auto it = peersCache_.begin();
            peer = it->second;
            id = it->first;
            peersCache_.erase(it);
        }
        peers_.emplace(std::make_pair(id, peer));
        return peer;
    }

    void
    deletePeer(Peer::id_t id, bool useCache = true)
    {
        auto it = peers_.find(id);
        assert(it != peers_.end());
        deletePeer(id, [&](PublicKey const&, PeerWPtr) {});
        if (useCache)
            peersCache_.emplace(std::make_pair(id, it->second));
        peers_.erase(it);
    }

    void
    resetPeers()
    {
        while (!peers_.empty())
            deletePeer(peers_.begin()->first);
        while (!peersCache_.empty())
            addPeer();
    }

    std::optional<Peer::id_t>
    deleteLastPeer()
    {
        if (peers_.empty())
            return {};

        std::uint8_t maxId = 0;

        for (auto& [id, _] : peers_)
        {
            (void)_;
            maxId = std::max<unsigned int>(id, maxId);
        }

        deletePeer(maxId, false);

        return maxId;
    }

    bool
    isCountingState(PublicKey const& validator)
    {
        return slots_.inState(validator, reduce_relay::SlotState::Counting);
    }

    std::set<id_t>
    getSelected(PublicKey const& validator)
    {
        return slots_.getSelected(validator);
    }

    bool
    isSelected(PublicKey const& validator, Peer::id_t peer)
    {
        auto selected = slots_.getSelected(validator);
        return selected.contains(peer);
    }

    id_t
    getSelectedPeer(PublicKey const& validator)
    {
        auto selected = slots_.getSelected(validator);
        assert(!selected.empty());
        return *selected.begin();
    }

    std::unordered_map<
        id_t,
        std::tuple<reduce_relay::PeerState, std::uint16_t, std::uint32_t, std::uint32_t>>
    getPeers(PublicKey const& validator)
    {
        return slots_.getPeers(validator);
    }

    std::uint16_t
    getNumPeers() const
    {
        return peers_.size();
    }

private:
    void
    squelch(PublicKey const& validator, Peer::id_t id, std::uint32_t squelchDuration) const override
    {
        if (auto it = peers_.find(id); it != peers_.end())
            squelch_(validator, it->second, squelchDuration);
    }
    void
    unsquelch(PublicKey const& validator, Peer::id_t id) const override
    {
        if (auto it = peers_.find(id); it != peers_.end())
            unsquelch_(validator, it->second);
    }
    SquelchCB squelch_;
    UnsquelchCB unsquelch_;
    Peers peers_;
    Peers peersCache_;
    reduce_relay::Slots<ManualClock> slots_;
    ServiceRegistry& registry_;
};

class Network
{
public:
    Network(Application& app) : overlay_(app)
    {
        init();
    }

    void
    init()
    {
        validators_.resize(kMaxValidators);
        for (int p = 0; p < kMaxPeers; p++)
        {
            auto peer = overlay_.addPeer();
            for (auto& v : validators_)
                v.addPeer(peer);
        }
    }

    ~Network() = default;

    void
    reset()
    {
        validators_.clear();
        overlay_.clear();
        PeerSim::resetId();
        Validator::resetId();
        init();
    }

    Peer::id_t
    addPeer()
    {
        auto peer = overlay_.addPeer();
        for (auto& v : validators_)
            v.addPeer(peer);
        return peer->id();
    }

    void
    deleteLastPeer()
    {
        auto id = overlay_.deleteLastPeer();

        if (!id)
            return;

        for (auto& validator : validators_)
            validator.deletePeer(*id);
    }

    void
    purgePeers()
    {
        while (overlay_.getNumPeers() > kMaxPeers)
            deleteLastPeer();
    }

    Validator&
    validator(std::uint16_t v)
    {
        assert(v < validators_.size());
        return validators_[v];
    }

    OverlaySim&
    overlay()
    {
        return overlay_;
    }

    void
    enableLink(std::uint16_t validatorId, Peer::id_t peer, bool enable)
    {
        auto it = std::ranges::find_if(validators_, [&](auto& v) { return v.id() == validatorId; });
        assert(it != validators_.end());
        if (enable)
        {
            it->linkUp(peer);
        }
        else
        {
            it->linkDown(peer);
        }
    }

    void
    onDisconnectPeer(Peer::id_t peer)
    {
        // Send unsquelch to the Peer on all links. This way when
        // the Peer "reconnects" it starts sending messages on the link.
        // We expect that if a Peer disconnects and then reconnects, it's
        // unsquelched.
        protocol::TMSquelch squelch;
        squelch.set_squelch(false);
        for (auto& v : validators_)
        {
            PublicKey const key = v;
            squelch.clear_validatorpubkey();
            squelch.set_validatorpubkey(key.data(), key.size());
            v.forLinks({peer}, [&](Link& l, MessageSPtr) {
                std::dynamic_pointer_cast<PeerSim>(l.getPeer())->send(squelch);
            });
        }
    }

    static void
    forRand(std::uint32_t min, std::uint32_t max, std::function<void(std::uint32_t)> f)
    {
        auto size = max - min;
        std::vector<std::uint32_t> s(size);
        std::iota(s.begin(), s.end(), min);  // NOLINT(modernize-use-ranges)
        std::random_device d;
        std::mt19937 g(d());
        std::shuffle(s.begin(), s.end(), g);
        for (auto v : s)
            f(v);
    }

    void
    propagate(
        LinkIterCB link,
        std::uint16_t nValidators = kMaxValidators,
        std::uint32_t nMessages = kMaxMessages,
        bool purge = true,
        bool resetClock = true)
    {
        if (resetClock)
            ManualClock::reset();

        if (purge)
        {
            purgePeers();
            overlay_.resetPeers();
        }

        for (int m = 0; m < nMessages; ++m)
        {
            ManualClock::randAdvance(milliseconds(1800), milliseconds(2200));
            forRand(0, nValidators, [&](std::uint32_t v) { validators_[v].forLinks(link); });
        }
    }

    /**
     * Is peer in Selected state in any of the slots
     */
    bool
    isSelected(Peer::id_t id)
    {
        return std::ranges::any_of(
            validators_, [&](auto& v) { return overlay_.isSelected(v, id); });
    }

    /**
     * Check if there are peers to unsquelch - peer is in Selected
     * state in any of the slots and there are peers in Squelched state
     * in those slots.
     */
    bool
    allCounting(Peer::id_t peer)
    {
        for (auto& v : validators_)
        {
            if (!overlay_.isSelected(v, peer))
                continue;
            auto peers = overlay_.getPeers(v);
            for (auto& [_, v] : peers)
            {
                (void)_;
                if (std::get<reduce_relay::PeerState>(v) == reduce_relay::PeerState::Squelched)
                    return false;
            }
        }
        return true;
    }

private:
    OverlaySim overlay_;
    std::vector<Validator> validators_;
};

class reduce_relay_test : public beast::unit_test::Suite
{
    using Slot = reduce_relay::Slot<ManualClock>;
    using id_t = Peer::id_t;

protected:
    void
    printPeers(std::string const& msg, std::uint16_t validator = 0)
    {
        auto peers = network_.overlay().getPeers(network_.validator(validator));
        std::cout << msg << " "
                  << "num peers " << (int)network_.overlay().getNumPeers() << std::endl;
        for (auto& [k, v] : peers)
            std::cout << k << ":" << (int)std::get<reduce_relay::PeerState>(v) << " ";
        std::cout << std::endl;
    }

    /**
     * Send squelch (if duration is set) or unsquelch (if duration not set)
     */
    static Peer::id_t
    sendSquelch(
        PublicKey const& validator,
        PeerWPtr const& peerPtr,
        std::optional<std::uint32_t> duration)
    {
        protocol::TMSquelch squelch;
        bool const res = static_cast<bool>(duration);
        squelch.set_squelch(res);
        squelch.set_validatorpubkey(validator.data(), validator.size());
        if (res)
            squelch.set_squelchduration(*duration);
        auto sp = peerPtr.lock();
        assert(sp);
        std::dynamic_pointer_cast<PeerSim>(sp)->send(squelch);
        return sp->id();
    }

    enum class State { On, Off, WaitReset };
    enum class EventType { LinkDown = 0, PeerDisconnected = 1 };
    // Link down or Peer disconnect event
    // TBD - add new peer event
    // TBD - add overlapping type of events at any
    //       time in any quantity
    struct Event
    {
        State state = State::Off;
        std::uint32_t cnt = 0;
        std::uint32_t handledCnt = 0;
        bool isSelected = false;
        Peer::id_t peer{};
        std::uint16_t validator{};
        std::optional<PublicKey> key;
        time_point<ManualClock> time;
        bool handled = false;
    };

    /**
     * Randomly brings the link between a validator and a peer down.
     * Randomly disconnects a peer. Those events are generated one at a time.
     */
    void
    random(bool log)
    {
        std::unordered_map<EventType, Event> events{
            {EventType::LinkDown, {}}, {EventType::PeerDisconnected, {}}};
        time_point<ManualClock> lastCheck = ManualClock::now();

        network_.reset();
        network_.propagate([&](Link& link, MessageSPtr m) {
            auto& validator = link.validator();
            auto now = ManualClock::now();

            bool squelched = false;
            std::stringstream str;

            link.send(
                m, [&](PublicKey const& key, PeerWPtr const& peerPtr, std::uint32_t duration) {
                    assert(key == validator);
                    auto p = sendSquelch(key, peerPtr, duration);
                    squelched = true;
                    str << p << " ";
                });

            if (squelched)
            {
                auto selected = network_.overlay().getSelected(validator);
                str << " selected: ";
                for (auto s : selected)
                    str << s << " ";
                if (log)
                {
                    std::cout << (double)reduce_relay::epoch<milliseconds>(now).count() / 1000.
                              << " random, squelched, validator: " << validator.id()
                              << " peers: " << str.str() << std::endl;
                }
                auto countingState = network_.overlay().isCountingState(validator);
                BEAST_EXPECT(
                    countingState == false &&
                    selected.size() == env_.app().config().vpReduceRelaySquelchMaxSelectedPeers);
            }

            // Trigger Link Down or Peer Disconnect event
            // Only one Link Down at a time
            if (events[EventType::LinkDown].state == State::Off)
            {
                auto update = [&](EventType event) {
                    events[event].cnt++;
                    events[event].validator = validator.id();
                    events[event].key = validator;
                    events[event].peer = link.peerId();
                    events[event].state = State::On;
                    events[event].time = now;
                    if (event == EventType::LinkDown)
                    {
                        network_.enableLink(validator.id(), link.peerId(), false);
                        events[event].isSelected =
                            network_.overlay().isSelected(validator, link.peerId());
                    }
                    else
                    {
                        events[event].isSelected = network_.isSelected(link.peerId());
                    }
                };
                auto r = randInt(0, 1000);
                if (r == (int)EventType::LinkDown || r == (int)EventType::PeerDisconnected)
                {
                    update(static_cast<EventType>(r));
                }
            }

            if (events[EventType::PeerDisconnected].state == State::On)
            {
                auto& event = events[EventType::PeerDisconnected];
                bool const allCounting = network_.allCounting(event.peer);
                network_.overlay().deletePeer(
                    event.peer, [&](PublicKey const& v, PeerWPtr const& peerPtr) {
                        if (event.isSelected)
                            sendSquelch(v, peerPtr, {});
                        event.handled = true;
                    });
                // Should only be unsquelched if the peer is in Selected state
                // If in Selected state it's possible unsquelching didn't
                // take place because there is no peers in Squelched state in
                // any of the slots where the peer is in Selected state
                // (allCounting is true)
                bool const handled = (!event.isSelected && !event.handled) ||
                    (event.isSelected && (event.handled || allCounting));
                BEAST_EXPECT(handled);
                event.state = State::Off;
                event.isSelected = false;
                event.handledCnt += handled;
                event.handled = false;
                network_.onDisconnectPeer(event.peer);
            }

            auto& event = events[EventType::LinkDown];
            // Check every sec for idled peers. Idled peers are
            // created by Link Down event.
            if (now - lastCheck > milliseconds(1000))
            {
                lastCheck = now;
                // Check if Link Down event must be handled by
                // deleteIdlePeer(): 1) the peer is in Selected state;
                // 2) the peer has not received any messages for IDLED time;
                // 3) there are peers in Squelched state in the slot.
                // 4) peer is in Slot's peers_ (if not then it is deleted
                //    by Slots::deleteIdlePeers())
                bool mustHandle = false;
                if (event.state == State::On && BEAST_EXPECT(event.key))
                {
                    event.isSelected = network_.overlay().isSelected(*event.key, event.peer);
                    auto peers = network_.overlay().getPeers(*event.key);
                    auto d = reduce_relay::epoch<milliseconds>(now).count() -
                        std::get<3>(peers[event.peer]);
                    mustHandle = event.isSelected &&
                        d > milliseconds(reduce_relay::kIdled).count() &&
                        network_.overlay().inState(*event.key, reduce_relay::PeerState::Squelched) >
                            0 &&
                        peers.contains(event.peer);
                }
                network_.overlay().deleteIdlePeers([&](PublicKey const& v, PeerWPtr const& ptr) {
                    event.handled = true;
                    if (mustHandle && v == event.key)
                    {
                        event.state = State::WaitReset;
                        sendSquelch(validator, ptr, {});
                    }
                });
                bool const handled = (event.handled && event.state == State::WaitReset) ||
                    (!event.handled && !mustHandle);
                BEAST_EXPECT(handled);
            }
            if (event.state == State::WaitReset ||
                (event.state == State::On &&
                 (now - event.time > (reduce_relay::kIdled + seconds(2)))))
            {
                bool const handled = event.state == State::WaitReset || !event.handled;
                BEAST_EXPECT(handled);
                event.state = State::Off;
                event.isSelected = false;
                event.handledCnt += handled;
                event.handled = false;
                network_.enableLink(event.validator, event.peer, true);
            }
        });

        auto& down = events[EventType::LinkDown];
        auto& disconnected = events[EventType::PeerDisconnected];
        // It's possible the last Down Link event is not handled
        BEAST_EXPECT(down.handledCnt >= down.cnt - 1);
        // All Peer Disconnect events must be handled
        BEAST_EXPECT(disconnected.cnt == disconnected.handledCnt);
        if (log)
        {
            std::cout << "link down count: " << down.cnt << "/" << down.handledCnt
                      << " peer disconnect count: " << disconnected.cnt << "/"
                      << disconnected.handledCnt;
        }
    }

    bool
    checkCounting(PublicKey const& validator, bool isCountingState)
    {
        auto countingState = network_.overlay().isCountingState(validator);
        BEAST_EXPECT(countingState == isCountingState);
        return countingState == isCountingState;
    }

    void
    doTest(std::string const& msg, bool log, std::function<void(bool)> f)
    {
        testcase(msg);
        f(log);
    }

    /**
     * Initial counting round: three peers receive message "faster" then
     * others. Once the message count for the three peers reaches threshold
     * the rest of the peers are squelched and the slot for the given validator
     * is in Selected state.
     */
    void
    testInitialRound(bool log)
    {
        doTest("Initial Round", log, [this](bool log) { BEAST_EXPECT(propagateAndSquelch(log)); });
    }

    /**
     * Receiving message from squelched peer too soon should not change the
     * slot's state to Counting.
     */
    void
    testPeerUnsquelchedTooSoon(bool log)
    {
        doTest("Peer Unsquelched Too Soon", log, [this](bool log) {
            BEAST_EXPECT(propagateNoSquelch(log, 1, false, false, false));
        });
    }

    /**
     * Receiving message from squelched peer should change the
     * slot's state to Counting.
     */
    void
    testPeerUnsquelched(bool log)
    {
        ManualClock::advance(seconds(601));
        doTest("Peer Unsquelched", log, [this](bool log) {
            BEAST_EXPECT(propagateNoSquelch(log, 2, true, true, false));
        });
    }

    /**
     * Propagate enough messages to generate one squelch event
     */
    bool
    propagateAndSquelch(bool log, bool purge = true, bool resetClock = true)
    {
        int n = 0;
        network_.propagate(
            [&](Link& link, MessageSPtr message) {
                std::uint16_t squelched = 0;
                link.send(
                    message,
                    [&](PublicKey const& key, PeerWPtr const& peerPtr, std::uint32_t duration) {
                        squelched++;
                        sendSquelch(key, peerPtr, duration);
                    });
                if (squelched)
                {
                    BEAST_EXPECT(
                        squelched ==
                        kMaxPeers - env_.app().config().vpReduceRelaySquelchMaxSelectedPeers);
                    n++;
                }
            },
            1,
            reduce_relay::kMaxMessageThreshold + 2,
            purge,
            resetClock);
        auto selected = network_.overlay().getSelected(network_.validator(0));
        BEAST_EXPECT(selected.size() == env_.app().config().vpReduceRelaySquelchMaxSelectedPeers);
        BEAST_EXPECT(n == 1);  // only one selection round
        auto res = checkCounting(network_.validator(0), false);
        BEAST_EXPECT(res);
        return n == 1 && res;
    }

    /**
     * Send fewer message so that squelch event is not generated
     */
    bool
    propagateNoSquelch(
        bool log,
        std::uint16_t nMessages,
        bool countingState,
        bool purge = true,
        bool resetClock = true)
    {
        bool squelched = false;
        network_.propagate(
            [&](Link& link, MessageSPtr message) {
                link.send(
                    message,
                    [&](PublicKey const& key, PeerWPtr const& peerPtr, std::uint32_t duration) {
                        squelched = true;
                        BEAST_EXPECT(false);
                    });
            },
            1,
            nMessages,
            purge,
            resetClock);
        auto res = checkCounting(network_.validator(0), countingState);
        return !squelched && res;
    }

    /**
     * Receiving a message from new peer should change the
     * slot's state to Counting.
     */
    void
    testNewPeer(bool log)
    {
        doTest("New Peer", log, [this](bool log) {
            BEAST_EXPECT(propagateAndSquelch(log, true, false));
            network_.addPeer();
            BEAST_EXPECT(propagateNoSquelch(log, 1, true, false, false));
        });
    }

    /**
     * Selected peer disconnects. Should change the state to counting and
     * unsquelch squelched peers.
     */
    void
    testSelectedPeerDisconnects(bool log)
    {
        doTest("Selected Peer Disconnects", log, [this](bool log) {
            ManualClock::advance(seconds(601));
            BEAST_EXPECT(propagateAndSquelch(log, true, false));
            auto id = network_.overlay().getSelectedPeer(network_.validator(0));
            std::uint16_t unsquelched = 0;
            network_.overlay().deletePeer(
                id, [&](PublicKey const& key, PeerWPtr const& peer) { unsquelched++; });
            BEAST_EXPECT(
                unsquelched ==
                kMaxPeers - env_.app().config().vpReduceRelaySquelchMaxSelectedPeers);
            BEAST_EXPECT(checkCounting(network_.validator(0), true));
        });
    }

    /**
     * Selected peer stops relaying. Should change the state to counting and
     * unsquelch squelched peers.
     */
    void
    testSelectedPeerStopsRelaying(bool log)
    {
        doTest("Selected Peer Stops Relaying", log, [this](bool log) {
            ManualClock::advance(seconds(601));
            BEAST_EXPECT(propagateAndSquelch(log, true, false));
            ManualClock::advance(reduce_relay::kIdled + seconds(1));
            std::uint16_t unsquelched = 0;
            network_.overlay().deleteIdlePeers(
                [&](PublicKey const& key, PeerWPtr const& peer) { unsquelched++; });
            auto peers = network_.overlay().getPeers(network_.validator(0));
            BEAST_EXPECT(
                unsquelched ==
                kMaxPeers - env_.app().config().vpReduceRelaySquelchMaxSelectedPeers);
            BEAST_EXPECT(checkCounting(network_.validator(0), true));
        });
    }

    /**
     * Squelched peer disconnects. Should not change the state to counting.
     */
    void
    testSquelchedPeerDisconnects(bool log)
    {
        doTest("Squelched Peer Disconnects", log, [this](bool log) {
            ManualClock::advance(seconds(601));
            BEAST_EXPECT(propagateAndSquelch(log, true, false));
            auto peers = network_.overlay().getPeers(network_.validator(0));
            auto it = std::ranges::find_if(peers, [&](auto it) {
                return std::get<reduce_relay::PeerState>(it.second) ==
                    reduce_relay::PeerState::Squelched;
            });
            assert(it != peers.end());
            std::uint16_t unsquelched = 0;
            network_.overlay().deletePeer(
                it->first, [&](PublicKey const& key, PeerWPtr const& peer) { unsquelched++; });
            BEAST_EXPECT(unsquelched == 0);
            BEAST_EXPECT(checkCounting(network_.validator(0), false));
        });
    }

    void
    testConfig(bool log)
    {
        doTest("Test Config - squelch enabled (legacy)", log, [&](bool log) {
            Config c;

            std::string const toLoad(R"xrpldConfig(
[reduce_relay]
vp_enable=1
)xrpldConfig");

            c.loadFromString(toLoad);
            BEAST_EXPECT(c.vpReduceRelayBaseSquelchEnable == true);
        });

        doTest("Test Config - squelch disabled (legacy)", log, [&](bool log) {
            Config c;

            std::string toLoad(R"xrpldConfig(
[reduce_relay]
vp_enable=0
)xrpldConfig");

            c.loadFromString(toLoad);
            BEAST_EXPECT(c.vpReduceRelayBaseSquelchEnable == false);

            Config c1;

            toLoad = R"xrpldConfig(
[reduce_relay]
)xrpldConfig";

            c1.loadFromString(toLoad);
            BEAST_EXPECT(c1.vpReduceRelayBaseSquelchEnable == false);
        });

        doTest("Test Config - squelch enabled", log, [&](bool log) {
            Config c;

            std::string const toLoad(R"xrpldConfig(
[reduce_relay]
vp_base_squelch_enable=1
)xrpldConfig");

            c.loadFromString(toLoad);
            BEAST_EXPECT(c.vpReduceRelayBaseSquelchEnable == true);
        });

        doTest("Test Config - squelch disabled", log, [&](bool log) {
            Config c;

            std::string const toLoad(R"xrpldConfig(
[reduce_relay]
vp_base_squelch_enable=0
)xrpldConfig");

            c.loadFromString(toLoad);
            BEAST_EXPECT(c.vpReduceRelayBaseSquelchEnable == false);
        });

        doTest("Test Config - legacy and new", log, [&](bool log) {
            Config c;

            std::string const toLoad(R"xrpldConfig(
[reduce_relay]
vp_base_squelch_enable=0
vp_enable=0
)xrpldConfig");

            std::string error;
            auto const expectedError =
                "Invalid reduce_relay"
                " cannot specify both vp_base_squelch_enable and vp_enable "
                "options. "
                "vp_enable was deprecated and replaced by "
                "vp_base_squelch_enable";

            try
            {
                c.loadFromString(toLoad);
            }
            catch (std::runtime_error const& e)
            {
                error = e.what();
            }

            BEAST_EXPECT(error == expectedError);
        });

        doTest("Test Config - max selected peers", log, [&](bool log) {
            Config c;

            std::string toLoad(R"xrpldConfig(
[reduce_relay]
)xrpldConfig");

            c.loadFromString(toLoad);
            BEAST_EXPECT(c.vpReduceRelaySquelchMaxSelectedPeers == 5);

            Config c1;

            toLoad = R"xrpldConfig(
[reduce_relay]
vp_base_squelch_max_selected_peers=6
)xrpldConfig";

            c1.loadFromString(toLoad);
            BEAST_EXPECT(c1.vpReduceRelaySquelchMaxSelectedPeers == 6);

            Config c2;

            toLoad = R"xrpldConfig(
[reduce_relay]
vp_base_squelch_max_selected_peers=2
)xrpldConfig";

            std::string error;
            auto const expectedError =
                "Invalid reduce_relay"
                " vp_base_squelch_max_selected_peers must be "
                "greater than or equal to 3";
            try
            {
                c2.loadFromString(toLoad);
            }
            catch (std::runtime_error const& e)
            {
                error = e.what();
            }

            BEAST_EXPECT(error == expectedError);
        });
    }

    void
    testBaseSquelchReady(bool log)
    {
        doTest("BaseSquelchReady", log, [&](bool log) {
            ManualClock::reset();
            auto createSlots = [&](bool baseSquelchEnabled) -> reduce_relay::Slots<ManualClock> {
                env_.app().config().vpReduceRelayBaseSquelchEnable = baseSquelchEnabled;
                return reduce_relay::Slots<ManualClock>(
                    env_.app(), network_.overlay(), env_.app().config());
            };
            // base squelching must not be ready if squelching is disabled
            BEAST_EXPECT(!createSlots(false).baseSquelchReady());

            // base squelch must not be ready as not enough time passed from
            // bootup
            BEAST_EXPECT(!createSlots(true).baseSquelchReady());

            ManualClock::advance(reduce_relay::kWaitOnBootup + minutes{1});

            // base squelch enabled and bootup time passed
            BEAST_EXPECT(createSlots(true).baseSquelchReady());

            // even if time passed, base squelching must not be ready if turned
            // off in the config
            BEAST_EXPECT(!createSlots(false).baseSquelchReady());
        });
    }

    void
    testInternalHashRouter(bool log)
    {
        doTest("Duplicate Message", log, [&](bool log) {
            network_.reset();
            // update message count for the same peer/validator
            std::int16_t const nMessages = 5;
            for (int i = 0; i < nMessages; i++)
            {
                uint256 const key(i);
                network_.overlay().updateSlotAndSquelch(
                    key, network_.validator(0), 0, [&](PublicKey const&, PeerWPtr, std::uint32_t) {
                    });
            }
            auto peers = network_.overlay().getPeers(network_.validator(0));
            // first message changes Slot state to Counting and is not counted,
            // hence '-1'.
            BEAST_EXPECT(std::get<1>(peers[0]) == (nMessages - 1));
            // add duplicate
            uint256 const key(nMessages - 1);
            network_.overlay().updateSlotAndSquelch(
                key, network_.validator(0), 0, [&](PublicKey const&, PeerWPtr, std::uint32_t) {});
            // confirm the same number of messages
            peers = network_.overlay().getPeers(network_.validator(0));
            BEAST_EXPECT(std::get<1>(peers[0]) == (nMessages - 1));
            // advance the clock
            ManualClock::advance(reduce_relay::kIdled + seconds(1));
            network_.overlay().updateSlotAndSquelch(
                key, network_.validator(0), 0, [&](PublicKey const&, PeerWPtr, std::uint32_t) {});
            peers = network_.overlay().getPeers(network_.validator(0));
            // confirm message number increased
            BEAST_EXPECT(std::get<1>(peers[0]) == nMessages);
        });
    }

    struct Handler : public reduce_relay::SquelchHandler
    {
        Handler() = default;
        void
        squelch(PublicKey const&, Peer::id_t, std::uint32_t duration) const override
        {
            maxDuration = std::max<uint32_t>(duration, maxDuration);
        }
        void
        unsquelch(PublicKey const&, Peer::id_t) const override
        {
        }
        mutable int maxDuration{0};
    };

    void
    testRandomSquelch(bool l)
    {
        doTest("Random Squelch", l, [&](bool l) {
            PublicKey validator = std::get<0>(randomKeyPair(KeyType::Ed25519));
            Handler handler;

            auto run = [&](int npeers) {
                handler.maxDuration = 0;
                reduce_relay::Slots<ManualClock> slots(env_.app(), handler, env_.app().config());
                // 1st message from a new peer switches the slot
                // to counting state and resets the counts of all peers +
                // MAX_MESSAGE_THRESHOLD + 1 messages to reach the threshold
                // and switch the slot's state to peer selection.
                for (int m = 1; m <= reduce_relay::kMaxMessageThreshold + 2; m++)
                {
                    for (int peer = 0; peer < npeers; peer++)
                    {
                        // make unique message hash to make the
                        // slot's internal hash router accept the message
                        std::uint64_t const mid = (m * 1000) + peer;
                        uint256 const message{mid};
                        slots.updateSlotAndSquelch(
                            message, validator, peer, protocol::MessageType::mtVALIDATION);
                    }
                }
                // make Slot's internal hash router expire all messages
                ManualClock::advance(hours(1));
            };

            using namespace reduce_relay;
            // expect max duration less than kMaxUnsquelchExpireDefault with
            // less than or equal to 60 peers
            run(20);
            BEAST_EXPECT(
                handler.maxDuration >= kMinUnsquelchExpire.count() &&
                handler.maxDuration <= kMaxUnsquelchExpireDefault.count());
            run(60);
            BEAST_EXPECT(
                handler.maxDuration >= kMinUnsquelchExpire.count() &&
                handler.maxDuration <= kMaxUnsquelchExpireDefault.count());
            // expect max duration greater than kMinUnsquelchExpire and less
            // than kMaxUnsquelchExpirePeers with peers greater than 60
            // and less than 360
            run(350);
            // can't make this condition stronger. squelch
            // duration is probabilistic and max condition may still fail.
            // log when the value is low
            BEAST_EXPECT(
                handler.maxDuration >= kMinUnsquelchExpire.count() &&
                handler.maxDuration <= kMaxUnsquelchExpirePeers.count());
            using namespace beast::unit_test::detail;
            if (handler.maxDuration <= kMaxUnsquelchExpireDefault.count())
            {
                log << makeReason("warning: squelch duration is low", __FILE__, __LINE__)
                    << std::endl
                    << std::flush;
            }
            // more than 400 is still less than kMaxUnsquelchExpirePeers
            run(400);
            BEAST_EXPECT(
                handler.maxDuration >= kMinUnsquelchExpire.count() &&
                handler.maxDuration <= kMaxUnsquelchExpirePeers.count());
            if (handler.maxDuration <= kMaxUnsquelchExpireDefault.count())
            {
                log << makeReason("warning: squelch duration is low", __FILE__, __LINE__)
                    << std::endl
                    << std::flush;
            }
        });
    }

    void
    testHandshake(bool log)
    {
        doTest("Handshake", log, [&](bool log) {
            auto setEnv = [&](bool enable) {
                Config c;
                std::stringstream str;
                str << "[reduce_relay]\n"
                    << "vp_enable=" << enable << "\n"
                    << "[compression]\n"
                    << "1\n";
                c.loadFromString(str.str());
                env_.app().config().vpReduceRelayBaseSquelchEnable =
                    c.vpReduceRelayBaseSquelchEnable;

                env_.app().config().compression = c.compression;
            };
            auto handshake = [&](int outboundEnable, int inboundEnable) {
                beast::IP::Address const addr = boost::asio::ip::make_address("172.1.1.100");

                setEnv(outboundEnable);
                auto request = xrpl::makeRequest(
                    true,
                    env_.app().config().compression,
                    false,
                    env_.app().config().txReduceRelayEnable,
                    env_.app().config().vpReduceRelayBaseSquelchEnable);
                http_request_type httpRequest;
                httpRequest.version(request.version());
                httpRequest.base() = request.base();
                // feature enabled on the peer's connection only if both sides
                // are enabled
                auto const peerEnabled = inboundEnable && outboundEnable;
                // inbound is enabled if the request's header has the feature
                // enabled and the peer's configuration is enabled
                auto const inboundEnabled =
                    peerFeatureEnabled(httpRequest, kFeatureVprr, inboundEnable);
                BEAST_EXPECT(!(peerEnabled ^ inboundEnabled));

                setEnv(inboundEnable);
                auto httpResp = xrpl::makeResponse(
                    true, httpRequest, addr, addr, uint256{1}, 1, {1, 0}, env_.app());
                // outbound is enabled if the response's header has the feature
                // enabled and the peer's configuration is enabled
                auto const outboundEnabled =
                    peerFeatureEnabled(httpResp, kFeatureVprr, outboundEnable);
                BEAST_EXPECT(!(peerEnabled ^ outboundEnabled));
            };
            handshake(1, 1);
            handshake(1, 0);
            handshake(0, 1);
            handshake(0, 0);
        });
    }

    jtx::Env env_;
    Network network_;

public:
    reduce_relay_test()
        : env_(*this, jtx::envconfig([](std::unique_ptr<Config> cfg) {
            cfg->vpReduceRelayBaseSquelchEnable = true;
            cfg->vpReduceRelaySquelchMaxSelectedPeers = 6;
            return cfg;
        }))
        , network_(env_.app())
    {
    }

    void
    run() override
    {
        bool const log = false;
        testConfig(log);
        testInitialRound(log);
        testPeerUnsquelchedTooSoon(log);
        testPeerUnsquelched(log);
        testNewPeer(log);
        testSquelchedPeerDisconnects(log);
        testSelectedPeerDisconnects(log);
        testSelectedPeerStopsRelaying(log);
        testInternalHashRouter(log);
        testRandomSquelch(log);
        testHandshake(log);
        testBaseSquelchReady(log);
    }
};

class reduce_relay_simulate_test : public reduce_relay_test
{
    void
    testRandom(bool log)
    {
        doTest("Random Test", log, [&](bool log) { random(log); });
    }

    void
    run() override
    {
        bool const log = false;
        testRandom(log);
    }
};

BEAST_DEFINE_TESTSUITE(reduce_relay, overlay, xrpl);
BEAST_DEFINE_TESTSUITE_MANUAL(reduce_relay_simulate, overlay, xrpl);

}  // namespace xrpl::test