#include "zeroerr/assert.h"
#include "zeroerr/unittest.h"

#include "data/ByteData.h"
#include "network/Channel.h"
#include "network/HttpRequest.h"
#include "network/NetHost.h"
#include "network/NetRpc.h"
#include "network/NetStream.h"
#include "network/NetWorker.h"
#include "network/Network.h"
#include "network/Session.h"
#include "network/TcpSocket.h"
#include "network/UdpLink.h"
#include "network/UdpSocket.h"
#include "platform_event/PlatformEvent.h"

#include <atomic>
#include <chrono>
#include <cstdio>
#include <cstring>
#include <string>
#include <thread>

using eve::platform_event::Message;
using eve::platform_event::PlatformEvent;
using eve::platform_event::Variant;

TEST_CASE("event.VariantMessage") {
    auto*                ev = eve::platform_event::PlatformEvent::create();
    std::vector<Variant> args;
    args.push_back(Variant::makeInt(200));
    args.push_back(Variant::makeString("ok"));
    args.push_back(Variant::makePtr(nullptr));
    ev->push(std::make_unique<Message>("httpresp", args));
    auto m = ev->pollOwned();
    REQUIRE(m.get() != nullptr);
    CHECK(m->name == "httpresp");
    REQUIRE(m->args.size() == 3);
    CHECK(static_cast<int>(m->args[0].type) == static_cast<int>(Variant::Type::Int));
    CHECK(m->args[0].i == 200);
    CHECK(m->args[1].s == "ok");
}

TEST_CASE("network.NetWorkerQueue") {
    eve::network::NetWorker worker(nullptr);
    eve::network::NetCompletion c;
    c.type   = eve::network::NetEvType::Err;
    c.reason = "timeout";
    worker.post(std::move(c));
    std::vector<eve::network::NetCompletion> out;
    worker.drain(out);
    REQUIRE(out.size() == 1);
    CHECK(out[0].reason == "timeout");
}

TEST_CASE("network.NetWorker.stopFromJobDoesNotDeadlock") {
    eve::network::NetWorker worker(nullptr);
    std::atomic<bool> stopped{false};
    worker.start();
    worker.submit([&] {
        worker.stop();
        stopped = true;
    });
    for (int i = 0; i < 200 && !stopped; ++i)
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
    CHECK(stopped.load());
    worker.stop();
}

TEST_CASE("network.Channel.fragmentedAndCoalescedFrames") {
    auto* net = eve::network::Network::create();
    auto*                   ev  = eve::platform_event::PlatformEvent::create();
    eve::network::TcpSocket socket(net);
    eve::network::Channel channel(&socket);

    channel.feed({0, 0});
    channel.feed({0, 2, 'a'});
    CHECK(!ev->pollOwned());
    channel.feed({'b', 0, 0, 0, 1, 'c'});

    auto first = ev->pollOwned();
    auto second = ev->pollOwned();
    REQUIRE(static_cast<bool>(first));
    REQUIRE(static_cast<bool>(second));
    CHECK(first->name == "chmsg");
    CHECK(second->name == "chmsg");
    REQUIRE(first->args.size() == 2);
    REQUIRE(second->args.size() == 2);
    auto* firstData = static_cast<eve::data::ByteData*>(first->args[1].p);
    auto* secondData = static_cast<eve::data::ByteData*>(second->args[1].p);
    REQUIRE(firstData != nullptr);
    REQUIRE(secondData != nullptr);
    CHECK(firstData->getSize() == 2);
    CHECK(secondData->getSize() == 1);
}

TEST_CASE("network.Channel.rejectsOversizedFrameHeader") {
    auto* net = eve::network::Network::create();
    auto* ev  = eve::platform_event::PlatformEvent::create();
    while (ev->pollOwned()) {}
    eve::network::TcpSocket socket(net);
    eve::network::Channel channel(&socket);
    channel.feed({0x01, 0x00, 0x00, 0x01});
    net->pump();

    bool sawClose = false;
    bool sawError = false;
    while (auto message = ev->pollOwned()) {
        if (message->name == "chclose") sawClose = true;
        if (message->name == "neterr") sawError = true;
    }
    CHECK(sawClose);
    CHECK(sawError);
}

TEST_CASE("network.TcpEcho") {
    auto* net = eve::network::Network::create();
    auto* ev  = eve::platform_event::PlatformEvent::create();
    REQUIRE(net != nullptr);
    REQUIRE(ev != nullptr);

    const uint16_t port = 39281;
    auto* server = net->newTcp();
    REQUIRE(server->listen(port));

    auto* client = net->newTcp();
    REQUIRE(client->connect("127.0.0.1", port));

    bool clientOk = false;
    eve::network::TcpSocket* peer = nullptr;
    for (int i = 0; i < 400 && (!clientOk || !peer); ++i) {
        net->pump();
        while (auto m = ev->pollOwned()) {
            if (m->name == "netconn" && m->args.size() >= 2 && m->args[1].s == "ok") {
                if (m->args[0].p == client) clientOk = true;
            }
        }
        if (!peer) peer = server->accept();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    REQUIRE(clientOk);
    REQUIRE(peer != nullptr);

    eve::data::ByteData payload("ping", 4);
    REQUIRE(client->send(&payload));

    bool got = false;
    for (int i = 0; i < 400 && !got; ++i) {
        net->pump();
        while (auto m = ev->pollOwned()) {
            if (m->name == "netdata") {
                got = true;
                REQUIRE(m->args.size() >= 2);
                CHECK(m->args[1].ownsPointer());
            }
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    CHECK(got);

    client->close();
    peer->close();
    server->close();
    delete client;
    delete peer;
    delete server;
}

TEST_CASE("network.UdpSendTo") {
    auto* net = eve::network::Network::create();
    auto* ev  = eve::platform_event::PlatformEvent::create();
    auto* a = net->newUdp();
    auto* b = net->newUdp();
    REQUIRE(a->bind(39301));
    REQUIRE(b->bind(39302));
    eve::data::ByteData payload("udp!", 4);
    REQUIRE(a->sendTo(&payload, "127.0.0.1", 39302));

    bool got = false;
    for (int i = 0; i < 400 && !got; ++i) {
        net->pump();
        while (auto m = ev->pollOwned()) {
            if (m->name == "netdata" && m->args.size() >= 1 && m->args[0].p == b) {
                got = true;
                REQUIRE(m->args.size() >= 2);
                CHECK(m->args[1].ownsPointer());
            }
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    CHECK(got);
    a->close();
    b->close();
    delete a;
    delete b;
}

TEST_CASE("network.HttpGetLocal") {
    auto* net = eve::network::Network::create();
    auto*          ev     = eve::platform_event::PlatformEvent::create();
    const uint16_t port = 39311;
    auto* server = net->newTcp();
    REQUIRE(server->listen(port));

    auto* req = net->newHttp("GET", "http://127.0.0.1:" + std::to_string(port) + "/");
    REQUIRE(req->submit());

    eve::network::TcpSocket* peer = nullptr;
    for (int i = 0; i < 400 && !peer; ++i) {
        net->pump();
        while (ev->pollOwned()) {}
        peer = server->accept();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    REQUIRE(peer != nullptr);

    const char* httpResp =
        "HTTP/1.1 200 OK\r\nContent-Length: 5\r\nConnection: close\r\n\r\nhello";
    eve::data::ByteData respData(httpResp, std::strlen(httpResp));
    REQUIRE(peer->send(&respData));

    bool got = false;
    int status = 0;
    for (int i = 0; i < 400 && !got; ++i) {
        net->pump();
        while (auto m = ev->pollOwned()) {
            if (m->name == "httpresp" && m->args.size() >= 2) {
                got = true;
                status = static_cast<int>(m->args[1].i);
                REQUIRE(m->args.size() >= 3);
                CHECK(m->args[2].ownsPointer());
            }
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    CHECK(got);
    CHECK(status == 200);

    peer->close();
    server->close();
    delete peer;
    delete server;
    delete req;
}

TEST_CASE("network.ChannelFrames") {
    auto* net = eve::network::Network::create();
    auto*          ev     = eve::platform_event::PlatformEvent::create();
    const uint16_t port = 39321;
    auto* server = net->newTcp();
    REQUIRE(server->listen(port));
    auto* client = net->newTcp();
    REQUIRE(client->connect("127.0.0.1", port));

    eve::network::TcpSocket* peer = nullptr;
    for (int i = 0; i < 400 && !peer; ++i) {
        net->pump();
        while (ev->pollOwned()) {}
        peer = server->accept();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    REQUIRE(peer != nullptr);

    auto* chSend = net->newChannel(client);
    auto* chRecv = net->newChannel(peer);
    REQUIRE(chSend->sendMsgString("ab"));
    REQUIRE(chSend->sendMsgString("cd"));

    int msgs = 0;
    for (int i = 0; i < 400 && msgs < 2; ++i) {
        net->pump();
        while (auto m = ev->pollOwned()) {
            if (m->name == "chmsg") {
                ++msgs;
                REQUIRE(m->args.size() >= 2);
                CHECK(m->args[1].ownsPointer());
            }
            if (m->name == "netdata" && m->args.size() >= 2 && m->args[1].p)
                CHECK(m->args[1].ownsPointer());
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    CHECK(msgs == 2);

    auto* sess = net->newSession();
    sess->add("main", chRecv);
    CHECK(sess->get("main") == chRecv);
    sess->closeAll();

    delete sess;
    delete chSend;
    delete chRecv;
    delete client;
    delete peer;
    delete server;
}

TEST_CASE("network.Network.timeoutAndVerifySsl") {
    auto* net = eve::network::Network::create();
    CHECK(net->getTimeout() == 10000);
    CHECK(net->getVerifySsl() == true);
    net->setTimeout(5000);
    net->setVerifySsl(false);
    CHECK(net->getTimeout() == 5000);
    CHECK(net->getVerifySsl() == false);
}

TEST_CASE("network.Network.cppFactoriesReturnOwners") {
    auto* net = eve::network::Network::create();
    auto tcp = net->makeTcp();
    auto udp = net->makeUdp();
    auto request = net->makeHttp("GET", "http://127.0.0.1/");
    auto session = net->makeSession();
    auto writer = net->makeWriter();
    auto reader = net->makeReader("bytes");
    REQUIRE(static_cast<bool>(tcp));
    REQUIRE(static_cast<bool>(udp));
    REQUIRE(static_cast<bool>(request));
    REQUIRE(static_cast<bool>(session));
    REQUIRE(static_cast<bool>(writer));
    REQUIRE(static_cast<bool>(reader));
    CHECK(!net->makeChannel(nullptr));
    CHECK(!net->makeUdpLink(nullptr));
    CHECK(!net->makeRpc(nullptr));
    REQUIRE(static_cast<bool>(net->makeHost()));
}

TEST_CASE("network.Tcp.connectedAndLocalPort") {
    auto* net = eve::network::Network::create();
    auto*          ev     = eve::platform_event::PlatformEvent::create();
    const uint16_t port = 39291;
    auto* server = net->newTcp();
    REQUIRE(server->listen(port));
    CHECK(server->getLocalPort() == port);
    CHECK(!server->isConnected());

    auto* client = net->newTcp();
    REQUIRE(client->connect("127.0.0.1", port));

    bool clientOk = false;
    eve::network::TcpSocket* peer = nullptr;
    for (int i = 0; i < 400 && (!clientOk || !peer); ++i) {
        net->pump();
        while (auto m = ev->pollOwned()) {
            if (m->name == "netconn" && m->args.size() >= 2 && m->args[1].s == "ok") {
                if (m->args[0].p == client) clientOk = true;
            }
        }
        if (!peer) peer = server->accept();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    REQUIRE(clientOk);
    REQUIRE(peer != nullptr);
    CHECK(client->isConnected());
    CHECK(peer->isConnected());
    CHECK(client->getLocalPort() != 0);
    CHECK(peer->getLocalPort() != 0);

    client->close();
    peer->close();
    server->close();
    delete client;
    delete peer;
    delete server;
}

TEST_CASE("network.Session.addGetRemoveCloseAll") {
    auto* net = eve::network::Network::create();
    auto*          ev     = eve::platform_event::PlatformEvent::create();
    const uint16_t port = 39331;
    auto* server = net->newTcp();
    REQUIRE(server->listen(port));
    auto* client = net->newTcp();
    REQUIRE(client->connect("127.0.0.1", port));

    eve::network::TcpSocket* peer = nullptr;
    for (int i = 0; i < 400 && !peer; ++i) {
        net->pump();
        while (ev->pollOwned()) {}
        peer = server->accept();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    REQUIRE(peer != nullptr);

    auto* chClient = net->newChannel(client);
    auto* chServer = net->newChannel(peer);
    auto* sess = net->newSession();

    sess->add("client", chClient);
    sess->add("server", chServer);
    CHECK(sess->get("client") == chClient);
    CHECK(sess->get("server") == chServer);
    CHECK(sess->get("missing") == nullptr);

    sess->remove("client");
    CHECK(sess->get("client") == nullptr);
    CHECK(sess->get("server") == chServer);

    sess->closeAll();
    CHECK(sess->get("server") == nullptr);

    delete sess;
    delete chClient;
    delete chServer;
    delete client;
    delete peer;
    delete server;
}

TEST_CASE("network.Http.setHeaderAndBody") {
    auto* net = eve::network::Network::create();
    auto*          ev     = eve::platform_event::PlatformEvent::create();
    const uint16_t port = 39341;
    auto* server = net->newTcp();
    REQUIRE(server->listen(port));

    auto* req = net->newHttp("POST", "http://127.0.0.1:" + std::to_string(port) + "/test");
    req->setHeader("X-Test-Header", "custom-value");
    req->setBodyString("post-body");
    REQUIRE(req->submit());

    std::string received;
    eve::network::TcpSocket* peer = nullptr;
    for (int i = 0; i < 400 && !peer; ++i) {
        net->pump();
        while (ev->pollOwned()) {}
        peer = server->accept();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    REQUIRE(peer != nullptr);

    for (int i = 0; i < 400 && received.find("\r\n\r\n") == std::string::npos; ++i) {
        net->pump();
        while (auto m = ev->pollOwned()) {
            if (m->name == "netdata" && m->args.size() >= 2 && m->args[0].p == peer && m->args[1].p) {
                auto* data = static_cast<eve::data::ByteData*>(m->args[1].p);
                received.append(static_cast<char*>(data->getData()), data->getSize());
            }
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    CHECK(received.find("X-Test-Header: custom-value") != std::string::npos);
    CHECK(received.find("post-body") != std::string::npos);

    const char* httpResp =
        "HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok";
    eve::data::ByteData respData(httpResp, std::strlen(httpResp));
    REQUIRE(peer->send(&respData));

    bool got = false;
    for (int i = 0; i < 400 && !got; ++i) {
        net->pump();
        while (auto m = ev->pollOwned()) {
            if (m->name == "httpresp") {
                got = true;
                REQUIRE(m->args.size() >= 3);
                CHECK(m->args[2].ownsPointer());
            }
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    CHECK(got);

    peer->close();
    server->close();
    delete peer;
    delete server;
    delete req;
}

TEST_CASE("network.Http.httpsReturnsTlsWithoutSslBackend") {
    auto* net = eve::network::Network::create();
    REQUIRE(net != nullptr);
    net->setVerifySsl(false);
    auto* http = net->newHttp("GET", "https://example.com/");
    REQUIRE(http != nullptr);
    http->setVerifySsl(true);  // request-level override is stored / resolved in submit()
    REQUIRE(http->submit());

    auto* ev = eve::platform_event::PlatformEvent::create();
    REQUIRE(ev != nullptr);
    bool gotTls = false;
    for (int i = 0; i < 200 && !gotTls; ++i) {
        net->pump();
        while (auto msg = ev->pollOwned()) {
            if (msg->name == "neterr" && msg->args.size() >= 2 && msg->args[1].s == "tls")
                gotTls = true;
        }
        if (!gotTls)
            std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    CHECK(gotTls);
    delete http;
}

TEST_CASE("network.NetWriterReaderRoundtrip") {
    eve::network::NetWriter w;
    w.writeU8(0xAB);
    w.writeI8(-3);
    w.writeU16(0xBEEF);
    w.writeI16(-1234);
    w.writeU32(0xDEADBEEFu);
    w.writeI32(-70000);
    w.writeU64(0x0123456789ABCDEFull);
    w.writeI64(-1234567890123ll);
    w.writeF32(1.5f);
    w.writeF64(-2.25);
    w.writeBool(true);
    w.writeBool(false);
    w.writeString("hello");
    w.writeString(std::string("a\0b", 3));
    const char raw[] = {0x00, 0x01, '\xFE', '\xFF'};
    w.writeBytes(raw, sizeof(raw));

    REQUIRE(w.size() > 0);
    eve::network::NetReader r(w.data(), w.size());
    REQUIRE(r.ok());
    CHECK(r.u8() == 0xAB);
    CHECK(r.i8() == -3);
    CHECK(r.u16() == 0xBEEF);
    CHECK(r.i16() == -1234);
    CHECK(r.u32() == 0xDEADBEEFu);
    CHECK(r.i32() == -70000);
    CHECK(r.u64() == 0x0123456789ABCDEFull);
    CHECK(r.i64() == -1234567890123ll);
    CHECK(r.f32() == 1.5f);
    CHECK(r.f64() == -2.25);
    CHECK(r.b() == true);
    CHECK(r.b() == false);
    CHECK(r.str() == "hello");
    CHECK(r.str() == std::string("a\0b", 3));
    auto b = r.bytes(sizeof(raw));
    REQUIRE(b.size() == sizeof(raw));
    CHECK(b[0] == 0x00);
    CHECK(b[1] == 0x01);
    CHECK(b[2] == '\xFE');
    CHECK(b[3] == '\xFF');
    CHECK(r.remaining() == 0);
    CHECK(r.ok());
    CHECK(r.pos() == w.size());
}

TEST_CASE("network.NetReaderOverrunIsSticky") {
    eve::network::NetWriter w;
    w.writeU32(7);
    eve::network::NetReader r(w.data(), w.size());
    REQUIRE(r.ok());
    CHECK(r.u32() == 7);
    CHECK(r.u16() == 0);  // overrun
    CHECK(!r.ok());
    CHECK(r.u8() == 0);   // safe default after failure
    CHECK(r.remaining() == 0);
    CHECK(r.str().empty());

    eve::network::NetReader bad(nullptr, 4);
    CHECK(!bad.ok());

    eve::network::NetWriter w2;
    w2.writeString("abc");
    eve::network::NetReader r2(w2.data(), w2.size());
    CHECK(r2.str() == "abc");
    CHECK(r2.u32() == 0);  // overrun after payload
    CHECK(!r2.ok());
}

TEST_CASE("network.TcpLargeSendFlushesFully") {
    auto* net = eve::network::Network::create();
    auto*          ev     = eve::platform_event::PlatformEvent::create();
    const uint16_t port = 39351;
    auto* server = net->newTcp();
    REQUIRE(server->listen(port));
    auto* client = net->newTcp();
    REQUIRE(client->connect("127.0.0.1", port));

    bool clientOk = false;
    eve::network::TcpSocket* peer = nullptr;
    for (int i = 0; i < 400 && (!clientOk || !peer); ++i) {
        net->pump();
        while (auto m = ev->pollOwned()) {
            if (m->name == "netconn" && m->args.size() >= 2 && m->args[1].s == "ok") {
                if (m->args[0].p == client) clientOk = true;
            }
        }
        if (!peer) peer = server->accept();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    REQUIRE(clientOk);
    REQUIRE(peer != nullptr);

    const size_t total = 2 * 1024 * 1024;  // 2 MiB, larger than any single send buffer
    std::string payload(total, 'x');
    for (size_t i = 0; i < total; ++i) payload[i] = static_cast<char>('a' + (i % 26));
    size_t got = 0;
    bool mismatch = false;

    for (size_t off = 0; off < total; off += 64 * 1024) {
        // Respect the 1 MiB send-buffer cap: wait for the worker to drain.
        for (int stall = 0; stall < 2000; ++stall) {
            if (client->pendingSendBytes() <= 128 * 1024) break;
            net->pump();
            while (auto m = ev->pollOwned()) {
                if (m->name == "netdata" && m->args.size() >= 2 && m->args[0].p == peer &&
                    m->args[1].p) {
                    auto* d = static_cast<eve::data::ByteData*>(m->args[1].p);
                    const char* p = static_cast<const char*>(d->getData());
                    for (size_t j = 0; j < d->getSize(); ++j) {
                        if (p[j] != payload[got + j]) mismatch = true;
                    }
                    got += d->getSize();
                }
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(2));
        }
        eve::data::ByteData chunk(payload.data() + off, 64 * 1024);
        REQUIRE(client->send(&chunk));
    }

    for (int i = 0; i < 2000 && got < total && !mismatch; ++i) {
        net->pump();
        while (auto m = ev->pollOwned()) {
            if (m->name == "netdata" && m->args.size() >= 2 && m->args[0].p == peer &&
                m->args[1].p) {
                auto* d = static_cast<eve::data::ByteData*>(m->args[1].p);
                const char* p = static_cast<const char*>(d->getData());
                for (size_t j = 0; j < d->getSize(); ++j) {
                    if (p[j] != payload[got + j]) mismatch = true;
                }
                got += d->getSize();
            }
        }
        if (got < total) std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    CHECK(!mismatch);
    CHECK(got == total);
    CHECK(client->pendingSendBytes() == 0);

    client->close();
    peer->close();
    server->close();
    delete client;
    delete peer;
    delete server;
}

TEST_CASE("network.ChannelLargeMessageReassembles") {
    auto* net = eve::network::Network::create();
    auto*          ev     = eve::platform_event::PlatformEvent::create();
    const uint16_t port = 39361;
    auto* server = net->newTcp();
    REQUIRE(server->listen(port));
    auto* client = net->newTcp();
    REQUIRE(client->connect("127.0.0.1", port));

    eve::network::TcpSocket* peer = nullptr;
    for (int i = 0; i < 400 && !peer; ++i) {
        net->pump();
        while (ev->pollOwned()) {}
        peer = server->accept();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    REQUIRE(peer != nullptr);

    auto* chSend = net->newChannel(client);
    auto* chRecv = net->newChannel(peer);
    REQUIRE(chRecv != nullptr);

    const size_t bigSize = 512 * 1024;
    std::string big(bigSize, 'z');
    REQUIRE(chSend->sendMsgString(big));

    std::string got;
    for (int i = 0; i < 4000 && got.size() < bigSize; ++i) {
        net->pump();
        while (auto m = ev->pollOwned()) {
            if (m->name == "chmsg" && m->args.size() >= 2 && m->args[1].p) {
                auto* d = static_cast<eve::data::ByteData*>(m->args[1].p);
                got.assign(static_cast<const char*>(d->getData()), d->getSize());
            }
        }
        if (got.size() < bigSize) std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    CHECK(got == big);

    delete chSend;
    delete chRecv;
    delete client;
    delete peer;
    delete server;
}

TEST_CASE("network.UdpReliableUnderLoss") {
    auto* net = eve::network::Network::create();
    auto* a = net->newUdp();
    auto* b = net->newUdp();
    REQUIRE(a->bind(39670));
    REQUIRE(b->bind(39671));
    auto* linkA = net->newUdpLink(a);
    auto* linkB = net->newUdpLink(b);
    REQUIRE(linkA->setRemote("127.0.0.1", 39671));
    REQUIRE(linkB->setRemote("127.0.0.1", 39670));
    linkA->setLossRate(0.25f);
    linkB->setLossRate(0.25f);

    std::vector<std::string> received;
    linkB->setMessageHandler(
        [&received](eve::network::UdpLink::MsgType, uint8_t, const char* d, size_t n) {
            received.emplace_back(d, n);
        });

    const int count = 40;
    for (int i = 0; i < count; ++i) {
        std::string msg = "m" + std::to_string(i) +
                          std::string(64 - std::to_string(i).size(), 'x');
        linkA->sendString(eve::network::UdpLink::MsgType::Reliable, 0, msg);
    }
    for (int i = 0; i < 2400 && received.size() < static_cast<size_t>(count); ++i) {
        net->pump();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    for (int i = 0; i < 600 && linkA->pendingReliable() > 0; ++i) {
        net->pump();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    CHECK(received.size() == static_cast<size_t>(count));
    CHECK(linkA->pendingReliable() == 0);
    bool inOrder = received.size() == static_cast<size_t>(count);
    for (int i = 0; i < count && inOrder; ++i) {
        std::string expect = "m" + std::to_string(i) +
                             std::string(64 - std::to_string(i).size(), 'x');
        if (received[static_cast<size_t>(i)] != expect) inOrder = false;
    }
    CHECK(inOrder);
}

TEST_CASE("network.UdpFragmentReassembly") {
    auto* net = eve::network::Network::create();
    auto* a = net->newUdp();
    auto* b = net->newUdp();
    REQUIRE(a->bind(39672));
    REQUIRE(b->bind(39673));
    auto* linkA = net->newUdpLink(a);
    auto* linkB = net->newUdpLink(b);
    REQUIRE(linkA->setRemote("127.0.0.1", 39673));
    REQUIRE(linkB->setRemote("127.0.0.1", 39672));
    linkA->setLossRate(0.1f);
    linkB->setLossRate(0.1f);

    std::string big(48 * 1024, 'q');
    for (size_t i = 0; i < big.size(); ++i) big[i] = static_cast<char>('a' + (i % 26));
    std::string got;
    linkB->setMessageHandler(
        [&got](eve::network::UdpLink::MsgType, uint8_t, const char* d, size_t n) {
            got.assign(d, n);
        });

    linkA->sendString(eve::network::UdpLink::MsgType::Reliable, 0, big);
    for (int i = 0; i < 2400 && got.size() < big.size(); ++i) {
        net->pump();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    for (int i = 0; i < 600 && linkA->pendingReliable() > 0; ++i) {
        net->pump();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    CHECK(got == big);
    CHECK(linkA->pendingReliable() == 0);
}

TEST_CASE("network.UdpUnreliableOrdered") {
    auto* net = eve::network::Network::create();
    auto* a = net->newUdp();
    auto* b = net->newUdp();
    REQUIRE(a->bind(39674));
    REQUIRE(b->bind(39675));
    auto* linkA = net->newUdpLink(a);
    auto* linkB = net->newUdpLink(b);
    REQUIRE(linkA->setRemote("127.0.0.1", 39675));
    REQUIRE(linkB->setRemote("127.0.0.1", 39674));

    std::vector<std::string> received;
    linkB->setMessageHandler(
        [&received](eve::network::UdpLink::MsgType, uint8_t, const char* d, size_t n) {
            received.emplace_back(d, n);
        });
    const int count = 12;
    for (int i = 0; i < count; ++i) {
        linkA->sendString(eve::network::UdpLink::MsgType::UnreliableOrdered, 1,
                          "o" + std::to_string(i));
    }
    for (int i = 0; i < 400 && received.size() < static_cast<size_t>(count); ++i) {
        net->pump();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    CHECK(received.size() == static_cast<size_t>(count));
    bool inOrder = received.size() == static_cast<size_t>(count);
    for (int i = 0; i < count && inOrder; ++i) {
        if (received[static_cast<size_t>(i)] != "o" + std::to_string(i)) inOrder = false;
    }
    CHECK(inOrder);
}

TEST_CASE("network.UdpRpcRoundtrip") {
    auto* net = eve::network::Network::create();
    auto* a = net->newUdp();
    auto* b = net->newUdp();
    REQUIRE(a->bind(39676));
    REQUIRE(b->bind(39677));
    auto* linkA = net->newUdpLink(a);
    auto* linkB = net->newUdpLink(b);
    REQUIRE(linkA->setRemote("127.0.0.1", 39677));
    REQUIRE(linkB->setRemote("127.0.0.1", 39676));
    auto* rpcA = net->newRpc(linkA);
    auto* rpcB = net->newRpc(linkB);
    REQUIRE(rpcA != nullptr);
    REQUIRE(rpcB != nullptr);

    std::string gotB, gotA;
    rpcB->registerHandler(7, [&gotB](eve::network::NetReader& r) { gotB = r.str(); });
    rpcA->registerHandler(8, [&gotA](eve::network::NetReader& r) { gotA = r.str(); });

    eve::network::NetWriter w;
    w.writeString("hello-from-a");
    rpcA->call(7, w.data(), w.size(), true);
    for (int i = 0; i < 600 && (gotB.empty() || gotA.empty()); ++i) {
        net->pump();
        if (!gotB.empty() && gotA.empty()) {
            eve::network::NetWriter w2;
            w2.writeString("reply-from-b");
            rpcB->call(8, w2.data(), w2.size(), true);
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    CHECK(gotB == "hello-from-a");
    CHECK(gotA == "reply-from-b");
}

TEST_CASE("network.NetHostPeers") {
    auto       phaseStart  = std::chrono::steady_clock::now();
    const auto recordPhase = [&](const char* phase) {
        const auto now = std::chrono::steady_clock::now();
        std::fprintf(stderr, "NetHostPeers %s: %.3f ms\n", phase,
                     std::chrono::duration<double, std::milli>(now - phaseStart).count());
        phaseStart = now;
    };
    auto*       net = eve::network::Network::create();
    std::string hostGot, clientGot;
    int connectedPeer = -1;
    int disconnectedPeer = -1;
    auto        host             = net->makeHost();
    REQUIRE(host->start(39680));
    host->setTimeoutMs(500);

    host->setMessageHandler(
        [&hostGot](uint32_t, eve::network::UdpLink::MsgType, uint8_t, const char* d,
                   size_t n) { hostGot.assign(d, n); });
    host->setPeerConnectedHandler([&connectedPeer](uint32_t id) {
        connectedPeer = static_cast<int>(id);
    });
    host->setPeerDisconnectedHandler([&disconnectedPeer](uint32_t id) {
        disconnectedPeer = static_cast<int>(id);
    });

    auto csock = net->makeUdp();
    REQUIRE(csock->connect("127.0.0.1", 39680));
    auto clink = net->makeUdpLink(csock.get());
    REQUIRE(clink->setRemote("127.0.0.1", 39680));
    clink->setTimeoutMs(500);
    clink->setMessageHandler(
        [&clientGot](eve::network::UdpLink::MsgType, uint8_t, const char* d, size_t n) {
            clientGot.assign(d, n);
        });

    recordPhase("setup");
    clink->sendString(eve::network::UdpLink::MsgType::Reliable, 0, "ping");
    for (int i = 0; i < 600 && (hostGot.empty() || connectedPeer < 0); ++i) {
        net->pump();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    recordPhase("receive ping");
    REQUIRE(hostGot == "ping");
    REQUIRE(connectedPeer == 1);
    REQUIRE(host->peerCount() == 1);

    host->sendStringTo(1, eve::network::UdpLink::MsgType::Reliable, 0, "pong");
    for (int i = 0; i < 600 && clientGot.empty(); ++i) {
        net->pump();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    recordPhase("receive pong");
    REQUIRE(clientGot == "pong");

    clink->setLossRate(1.0f);
    for (int i = 0; i < 600 && disconnectedPeer < 0; ++i) {
        net->pump();
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    recordPhase("loss and timeout");
    REQUIRE(disconnectedPeer == 1);
    REQUIRE(host->peerCount() == 0);
    clink.reset();
    csock.reset();
    host.reset();
    recordPhase("destroy client and host");
}