// Copyright (C) 2025 The Qt Company Ltd.
// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only

#include <mockserver.h>

#include <proto/server/event.pb.h>
#include <proto/server/eventhub.grpc.pb.h>

#include <proto/client/event.qpb.h>
#include <proto/client/eventhub_client.grpc.qpb.h>

#include <QtGrpc/qgrpccalloptions.h>
#include <QtGrpc/qgrpcchanneloptions.h>
#include <QtGrpc/qgrpchttp2channel.h>

#include <QtTest/qsignalspy.h>
#include <QtTest/qtest.h>

#include <QtCore/qbytearray.h>
#include <QtCore/qdatetime.h>
#include <QtCore/qhash.h>
#include <QtCore/qscopeguard.h>
#include <QtCore/qtimer.h>

#include <atomic>
#include <string>
#include <vector>

// Re-define so that we can use QTest Macros inside non void functions too.
#undef QTEST_FAIL_ACTION
#define QTEST_FAIL_ACTION                         \
    do {                                          \
        std::cerr << "Test failed!" << std::endl; \
        std::abort();                             \
    } while (0)

using namespace Qt::Literals::StringLiterals;
using namespace QtGrpc;
using MultiHash = QMultiHash<QByteArray, QByteArray>;

class QtGrpcClientEnd2EndTest : public QObject
{
    Q_OBJECT

public:
    static std::string serverHttpAddress() { return "localhost:50061"; }
    static std::string serverHttpsAddress() { return "localhost:50062"; }
    static std::string serverUnixAddress() { return "unix:///tmp/qtgrpc_test_end2end.sock"; }
    static std::string serverUnixAbstractAddress() { return "unix-abstract:qtgrpc_test_end2end"; }
    static std::vector<ListeningPort> serverListeningPorts()
    {
        return {
            { serverHttpAddress(),         grpc::InsecureServerCredentials() },
#if QT_CONFIG(ssl)
            { serverHttpsAddress(),        serverSslCredentials()            },
#endif
#ifdef Q_OS_UNIX
            { serverUnixAddress(),         grpc::InsecureServerCredentials() },
#endif

#ifdef Q_OS_LINUX
            { serverUnixAbstractAddress(), grpc::InsecureServerCredentials() },
#endif
        };
    }

private Q_SLOTS:
    void initTestCase_data() const;
    void initTestCase();
    void cleanupTestCase();

    void init();
    void cleanup();

    // Testcases:
    void clientMetadataReceived_data() const;
    void clientMetadataReceived();
    void serverMetadataReceived_data() const;
    void serverMetadataReceived();
    void serverInitialMetadataEmitted();

    void bidiStreamsInOrder();
    void hostUriRoundTrip();

    void clientHandlesCompression_data() const;
    void clientHandlesCompression();

    void channelChangeCancelsInFlightRPCs_data() const;
    void channelChangeCancelsInFlightRPCs();

    void maximumReceiveMessageSize_data() const;
    void maximumReceiveMessageSize();

    void acceptedCompressionAlgorithms_data() const;
    void acceptedCompressionAlgorithms();

    void requestCompression_data() const;
    void requestCompression();

private:
    static std::shared_ptr<grpc::ServerCredentials> serverSslCredentials()
    {
        grpc::SslServerCredentialsOptions opts(GRPC_SSL_DONT_REQUEST_CLIENT_CERTIFICATE);
        opts.pem_key_cert_pairs.push_back({ SslKey, SslCert });
        return grpc::SslServerCredentials(opts);
    }

    void restartServer(const grpc::ChannelArguments &channelArgs = {})
    {
        if (m_server)
            QVERIFY(m_server->stop());
        m_service = std::make_unique<EventHub::AsyncService>();
        m_server = std::make_unique<MockServer>();
        QVERIFY(m_server->start(serverListeningPorts(), { m_service.get() }, channelArgs));
    }

private:
    std::unique_ptr<MockServer> m_server;
    std::unique_ptr<EventHub::AsyncService> m_service;
    std::unique_ptr<qt::EventHub::Client> m_client;
};

void QtGrpcClientEnd2EndTest::initTestCase_data() const
{
    QTest::addColumn<QUrl>("hostUri");
    QTest::addColumn<QGrpcChannelOptions>("channelOptions");

    QTest::newRow("http") << QUrl("http://"_ba + QByteArrayView(serverHttpAddress()))
                          << QGrpcChannelOptions{};

#if QT_CONFIG(ssl)
    QSslConfiguration tlsConfig;
    tlsConfig.setProtocol(QSsl::TlsV1_2);
    tlsConfig.setCaCertificates({ QSslCertificate{ QByteArray(SslCert) } });
    tlsConfig.setAllowedNextProtocols({ "h2"_ba });
    QGrpcChannelOptions tlsOpts;
    tlsOpts.setSslConfiguration(tlsConfig);
    QTest::newRow("https") << QUrl("https://"_ba + QByteArrayView(serverHttpsAddress())) << tlsOpts;
#endif

#ifdef Q_OS_UNIX
    QTest::newRow("unix") << QUrl(serverUnixAddress().data()) << QGrpcChannelOptions{};
#endif

#ifdef Q_OS_LINUX
    QTest::newRow("unix-abstract")
        << QUrl(serverUnixAbstractAddress().data()) << QGrpcChannelOptions{};
#endif
}

void QtGrpcClientEnd2EndTest::initTestCase()
{
    QTest::failOnWarning();
    restartServer();
}

void QtGrpcClientEnd2EndTest::cleanupTestCase()
{
    m_client.reset();
    QVERIFY(m_server->stop());
    m_service.reset();
}

void QtGrpcClientEnd2EndTest::init()
{
    QVERIFY(m_service && m_server);
    QFETCH_GLOBAL(const QUrl, hostUri);
    QFETCH_GLOBAL(const QGrpcChannelOptions, channelOptions);
    m_client = std::make_unique<qt::EventHub::Client>();
    QVERIFY(m_client->attachChannel(std::make_shared<QGrpcHttp2Channel>(hostUri, channelOptions)));
}

void QtGrpcClientEnd2EndTest::cleanup()
{
    m_client.reset();
}

void QtGrpcClientEnd2EndTest::clientMetadataReceived_data() const
{
    QTest::addColumn<MultiHash>("callMetadata");
    QTest::addColumn<MultiHash>("channelMetadata");
    MultiHash callMd{
        { "client-call-single", "call-value-1" },
        { "client-call-multi",  "call-a"       },
        { "client-call-multi",  "call-b"       }
    };
    MultiHash channelMd{
        { "client-channel-single", "channel-value-1" },
        { "client-channel-multi",  "channel-a"       },
        { "client-channel-multi",  "channel-b"       }
    };
    QTest::addRow("call") << callMd << MultiHash{};
    QTest::addRow("channel") << MultiHash{} << channelMd;
    QTest::addRow("call+channel") << callMd << channelMd;
}

void QtGrpcClientEnd2EndTest::clientMetadataReceived()
{
    QFETCH(const MultiHash, callMetadata);
    QFETCH(const MultiHash, channelMetadata);

    // Setup Server-side handling
    auto processor = m_server->createProcessor();
    struct ServerData
    {
        grpc::ServerAsyncResponseWriter<None> op{ &ctx };
        grpc::ServerContext ctx;

        Event request;
        None response;
    };
    ServerData *data = new ServerData;

    CallbackTag *callHandler = new CallbackTag(
        [&](bool ok) {
            QVERIFY(ok);

            const std::multimap<grpc::string_ref, grpc::string_ref>
                &receivedMd = data->ctx.client_metadata();
            auto mergedMd = channelMetadata;
            mergedMd.unite(callMetadata);

            for (auto it = mergedMd.cbegin(); it != mergedMd.cend(); ++it) {
                // Check that each key-value pair sent by the client exists on the server
                auto serverRange = receivedMd.equal_range(it.key().toStdString());
                auto clientRange = mergedMd.equal_range(it.key());

                QCOMPARE_EQ(std::distance(serverRange.first, serverRange.second),
                            std::distance(clientRange.first, clientRange.second));
                while (clientRange.first != clientRange.second) {
                    // Look for the exact entry in the server range. The order may
                    // be changed but it must be present.
                    const auto it = std::find_if(serverRange.first, serverRange.second,
                                                 [&](auto it) {
                                                     return it.first
                                                         == clientRange.first.key().toStdString()
                                                         && it.second
                                                         == clientRange.first.value().toStdString();
                                                 });
                    QVERIFY(it != serverRange.second);
                    std::advance(clientRange.first, 1);
                }
            }
            data->op.Finish(data->response, grpc::Status::OK,
                            new DeleteTag<ServerData>(data, processor.get()));
            return CallbackTag::Delete;
        },
        processor.get());
    m_service->RequestPush(&data->ctx, &data->request, &data->op, m_server->cq(), m_server->cq(),
                           callHandler);

    // Setup Client-side call
    m_client->channel()->setChannelOptions(QGrpcChannelOptions().setMetadata(channelMetadata));
    auto call = m_client->Push(qt::Event{}, QGrpcCallOptions().setMetadata(callMetadata));
    QVERIFY(call);

    connect(call.get(), &QGrpcOperation::finished, this, [&](const QGrpcStatus &status) {
        QVERIFY(status.isOk());
        auto response = call->read<qt::None>();
        QVERIFY(response.has_value());
    });

    QSignalSpy finishedSpy(call.get(), &QGrpcOperation::finished);
    QVERIFY(finishedSpy.isValid());
    QVERIFY(finishedSpy.wait());
}

void QtGrpcClientEnd2EndTest::serverMetadataReceived_data() const
{
    QTest::addColumn<bool>("filterServerMetadata");
    QTest::addColumn<MultiHash>("expectedInitialMd");
    QTest::addColumn<MultiHash>("expectedTrailingMd");

    MultiHash initialMd{
        { "initial-1", "ivalue-1" },
        { "initial-2", "ivalue-2" }
    };
    MultiHash trailingMd{
        { "trailing-1",     "tvalue-1" },
        { "trailing-multi", "tvalue-x" },
        { "trailing-multi", "tvalue-y" }
    };

    QTest::addRow("filter(true)") << true << initialMd << trailingMd;
    QTest::addRow("filter(false)") << false << initialMd << trailingMd;
}

void QtGrpcClientEnd2EndTest::serverMetadataReceived()
{
    using MultiHash = QMultiHash<QByteArray, QByteArray>;
    QFETCH(const bool, filterServerMetadata);
    QFETCH(const MultiHash, expectedInitialMd);
    QFETCH(const MultiHash, expectedTrailingMd);

    // Setup Server-side handling
    auto processor = m_server->createProcessor();
    struct ServerData
    {
        grpc::ServerAsyncResponseWriter<None> op{ &ctx };
        grpc::ServerContext ctx;

        Event request;
        None response;
    };
    ServerData *data = new ServerData;

    for (auto it = expectedInitialMd.cbegin(); it != expectedInitialMd.cend(); ++it)
        data->ctx.AddInitialMetadata(it.key().toStdString(), it.value().toStdString());
    for (auto it = expectedTrailingMd.cbegin(); it != expectedTrailingMd.cend(); ++it)
        data->ctx.AddTrailingMetadata(it.key().toStdString(), it.value().toStdString());

    CallbackTag *callHandler = new CallbackTag(
        [&](bool ok) {
            QVERIFY(ok);
            data->op.Finish(data->response, grpc::Status::OK,
                            new DeleteTag<ServerData>(data, processor.get()));
            return CallbackTag::Delete;
        },
        processor.get());
    m_service->RequestPush(&data->ctx, &data->request, &data->op, m_server->cq(), m_server->cq(),
                           callHandler);

    // Setup Client-side call
    auto chOpts = QGrpcChannelOptions().setFilterServerMetadata(filterServerMetadata);
    m_client->channel()->setChannelOptions(chOpts);

    auto call = m_client->Push(qt::Event{});
    QVERIFY(call);

    connect(call.get(), &QGrpcOperation::finished, this, [&](const QGrpcStatus &status) {
        QVERIFY(status.isOk());
        auto response = call->read<qt::None>();
        QVERIFY(response.has_value());

        const auto &initialMd = call->serverInitialMetadata();
        const auto &trailingMd = call->serverTrailingMetadata();

        if (filterServerMetadata) {
            QCOMPARE(initialMd, expectedInitialMd);
            QCOMPARE(trailingMd, expectedTrailingMd);
        } else {
            QCOMPARE_GE(initialMd.size(), expectedInitialMd.size());
            QCOMPARE_GE(trailingMd.size(), expectedTrailingMd.size());
            for (auto it = expectedInitialMd.cbegin(); it != expectedInitialMd.cend(); ++it)
                QVERIFY(initialMd.contains(it.key(), it.value()));
            for (auto it = expectedTrailingMd.cbegin(); it != expectedTrailingMd.cend(); ++it)
                QVERIFY(trailingMd.contains(it.key(), it.value()));
        }
    });

    QSignalSpy finishedSpy(call.get(), &QGrpcOperation::finished);
    QVERIFY(finishedSpy.isValid());
    QVERIFY(finishedSpy.wait());
}

void QtGrpcClientEnd2EndTest::serverInitialMetadataEmitted()
{
    // Setup Server-side handling
    auto processor = m_server->createProcessor();
    struct ServerData
    {
        grpc::ServerAsyncResponseWriter<None> op{ &ctx };
        grpc::ServerContext ctx;

        Event request;
        None response;
    };
    ServerData *data = new ServerData;
    data->ctx.AddInitialMetadata("initial", "value");
    data->ctx.AddTrailingMetadata("trailing", "value");

    CallbackTag *callHandler = new CallbackTag(
        [&](bool ok) {
            QVERIFY(ok);
            data->op.SendInitialMetadata(new CallbackTag(
                [&](bool ok) {
                    QVERIFY(ok);
                    // Wait one second before emitting finished.
                    std::this_thread::sleep_for(std::chrono::seconds(1));
                    data->op.Finish(data->response, grpc::Status::OK,
                                    new DeleteTag<ServerData>(data, processor.get()));
                    return CallbackTag::Delete;
                },
                processor.get()));
            return CallbackTag::Delete;
        },
        processor.get());
    m_service->RequestPush(&data->ctx, &data->request, &data->op, m_server->cq(), m_server->cq(),
                           callHandler);

    // Setup Client-side call
    QDateTime initialMetadataTime;
    QDateTime finishedTime;
    auto call = m_client->Push(qt::Event{}, QGrpcCallOptions{}.setFilterServerMetadata(true));
    QVERIFY(call);

    connect(call.get(), &QGrpcOperation::finished, this, [&](const QGrpcStatus &status) {
        finishedTime = QDateTime::currentDateTime();
        QVERIFY(status.isOk());
        QCOMPARE_EQ(call->serverTrailingMetadata().size(), 1);
    });
    connect(call.get(), &QGrpcOperation::serverInitialMetadataReceived, this,
            [&](const QMultiHash<QByteArray, QByteArray> &metadata) {
                initialMetadataTime = QDateTime::currentDateTime();
                QCOMPARE_EQ(call->serverInitialMetadata().size(), 1);
                QCOMPARE_EQ(call->serverInitialMetadata(), metadata);
            });

    QSignalSpy finishedSpy(call.get(), &QGrpcOperation::finished);
    QVERIFY(finishedSpy.isValid());
    QSignalSpy initialMetadataSpy(call.get(), &QGrpcOperation::serverInitialMetadataReceived);
    QVERIFY(initialMetadataSpy.isValid());

    finishedSpy.wait();

    QCOMPARE_EQ(initialMetadataSpy.count(), 1);
    QCOMPARE_LT(initialMetadataTime, finishedTime);
}

void QtGrpcClientEnd2EndTest::bidiStreamsInOrder()
{
    constexpr auto SleepTime = std::chrono::milliseconds(5);

    // Setup Server-side handling
    auto processor = m_server->createProcessor();
    struct ServerData
    {
        grpc::ServerAsyncReaderWriter<Event, Event> op{ &ctx };
        grpc::ServerContext ctx;

        Event request;
        Event response;
        unsigned long count = 0;
        std::atomic<bool> readerDone = false;
        std::atomic<bool> writerDone = false;

        void updateResponse()
        {
            response.set_type(Event::SERVER);
            response.set_number(response.number() + 1);
            response.set_name("server-" + std::to_string(response.number()));
        }
    };
    ServerData *data = new ServerData;

    CallbackTag *reader = new CallbackTag(
        [&, current = 1u](bool ok) mutable {
            if (!ok) {
                data->readerDone = true;
                if (data->writerDone)
                    data->op.Finish(grpc::Status::OK,
                                    new DeleteTag<ServerData>(data, processor.get()));
                return CallbackTag::Delete;
            }
            QCOMPARE_EQ(data->request.type(), Event::CLIENT);
            QCOMPARE_EQ(data->request.number(), current);
            std::string name = "client-" + std::to_string(current);
            QCOMPARE_EQ(data->request.name(), name);
            ++current;

            data->op.Read(&data->request, reader);
            return CallbackTag::Proceed;
        },
        processor.get());
    CallbackTag *writer = new CallbackTag(
        [&](bool ok) {
            QVERIFY(ok);
            if (data->response.number() >= data->count) {
                data->writerDone = true;
                if (data->readerDone)
                    data->op.Finish(grpc::Status::OK,
                                    new DeleteTag<ServerData>(data, processor.get()));
                return CallbackTag::Delete;
            }
            std::this_thread::sleep_for(SleepTime);
            data->updateResponse();
            data->op.Write(data->response, writer);
            return CallbackTag::Proceed;
        },
        processor.get());
    CallbackTag *callHandler = new CallbackTag(
        [&](bool ok) {
            QVERIFY(ok);
            const auto &md = data->ctx.client_metadata();
            const auto countIt = md.find("call-count");
            QVERIFY(countIt != md.cend());
            data->count = std::stoul(std::string(countIt->second.data(), countIt->second.length()));
            QCOMPARE_GT(data->count, 0u);

            data->op.Read(&data->request, reader);
            data->updateResponse();
            data->op.Write(data->response, writer);

            return CallbackTag::Delete;
        },
        processor.get());
    m_service->RequestExchange(&data->ctx, &data->op, m_server->cq(), m_server->cq(), callHandler);

    // Client bidi stream
    uint callCount = 25;
    qt::Event request;

    auto updateRequest = [&] {
        request.setType(qt::Event::Type::CLIENT);
        request.setNumber(request.number() + 1);
        request.setName("client-"_L1 + QString::number(request.number()));
    };

    updateRequest();
    auto copts = QGrpcCallOptions().addMetadata("call-count", QByteArray::number(callCount));
    auto stream = m_client->Exchange(request, copts);
    QVERIFY(stream);

    connect(stream.get(), &QGrpcOperation::finished, this,
            [](const QGrpcStatus &status) { QVERIFY(status.isOk()); });
    connect(stream.get(), &QGrpcBidiStream::messageReceived, this, [&, current = 1u]() mutable {
        const auto response = stream->read<qt::Event>();
        QVERIFY(response.has_value());
        QCOMPARE_EQ(response->type(), qt::Event::Type::SERVER);
        QCOMPARE_EQ(response->number(), current);
        QString name = "server-"_L1 + QString::number(current);
        QCOMPARE_EQ(response->name(), name);
        ++current;
    });

    QTimer delayedWriter;
    connect(&delayedWriter, &QTimer::timeout, this, [&, current = 1u]() mutable {
        if (current >= callCount) {
            stream->writesDone();
            delayedWriter.stop();
        }
        updateRequest();
        stream->writeMessage(request);
        ++current;
    });
    delayedWriter.start(SleepTime);

    QSignalSpy finishedSpy(stream.get(), &QGrpcOperation::finished);
    QVERIFY(finishedSpy.isValid());
    QVERIFY(finishedSpy.wait());
}

void QtGrpcClientEnd2EndTest::hostUriRoundTrip()
{
    auto *initialChannel = static_cast<QGrpcHttp2Channel *>(m_client->channel().get());
    QVERIFY(initialChannel);

    const QUrl sanitizedUri = initialChannel->hostUri();
    auto rebuiltChannel = std::make_shared<QGrpcHttp2Channel>(sanitizedUri,
                                                              initialChannel->channelOptions());
    // sanitizeHostUri must be idempotent: feeding hostUri() back in yields the
    // same URI, which is what keeps unix/unix-abstract channels reconstructible.
    QCOMPARE(rebuiltChannel->hostUri(), sanitizedUri);
    QVERIFY(m_client->attachChannel(rebuiltChannel));

    auto processor = m_server->createProcessor();
    struct ServerData
    {
        grpc::ServerAsyncResponseWriter<None> op{ &ctx };
        grpc::ServerContext ctx;
        Event request;
        None response;
    };
    ServerData *data = new ServerData;
    CallbackTag *callHandler = new CallbackTag(
        [&, data](bool ok) {
            QVERIFY(ok);
            data->op.Finish(data->response, grpc::Status::OK,
                            new DeleteTag<ServerData>(data, processor.get()));
            return CallbackTag::Delete;
        },
        processor.get());
    m_service->RequestPush(&data->ctx, &data->request, &data->op, m_server->cq(), m_server->cq(),
                           callHandler);

    auto call = m_client->Push(qt::Event{});
    QVERIFY(call);
    QSignalSpy finishedSpy(call.get(), &QGrpcOperation::finished);
    QVERIFY(finishedSpy.isValid());
    QVERIFY(finishedSpy.wait());
    QCOMPARE_EQ(finishedSpy.count(), 1);
    QVERIFY(finishedSpy.takeFirst().at(0).value<QGrpcStatus>().isOk());
}

void QtGrpcClientEnd2EndTest::clientHandlesCompression_data() const
{
    QTest::addColumn<grpc_compression_algorithm>("compressionAlgo");
    QTest::addRow("compress(None)") << GRPC_COMPRESS_NONE;
    QTest::addRow("compress(Deflate)") << GRPC_COMPRESS_DEFLATE;
    QTest::addRow("compress(Gzip)") << GRPC_COMPRESS_GZIP;
}

void QtGrpcClientEnd2EndTest::clientHandlesCompression()
{
    QFETCH(const grpc_compression_algorithm, compressionAlgo);
    EventList serverResponses;

    class SubscribeListHandler : public AbstractRpcTag
    {
    public:
        SubscribeListHandler(EventList &responses_, EventHub::AsyncService &service_,
                             const grpc_compression_algorithm compressionAlgo_,
                             TagProcessor *processor)
            : AbstractRpcTag(processor), op(&context()), service(service_), responses(responses_),
              compressionAlgo(compressionAlgo_)
        {
            context().set_compression_algorithm(compressionAlgo);
            context().set_compression_level(GRPC_COMPRESS_LEVEL_HIGH);
            // create some 'compressable' data. Try to make it more complex
            // as compression is not guaranteed to actually be applied.
            for (size_t i = 0; i < 100; ++i) {
                const auto v = i % 10;
                Event ev;
                ev.set_name("server;server;" + std::to_string(v));
                ev.set_number(v);
                responses.mutable_events()->Add(std::move(ev));
            }
        }
        void start(grpc::ServerCompletionQueue *cq) override
        {
            service.RequestSubscribeList(&context(), &request, &op, cq, cq, this);
        }
        void process(bool ok) override
        {
            QVERIFY(ok);
            if (index >= responseCount) {
                op.Finish(grpc::Status::OK, new DeleteTag<SubscribeListHandler>(this, processor));
                return;
            }

            grpc::WriteOptions wopts;
            // Enable and disable the compression per-message
            if (index % 2 == 0)
                wopts.set_no_compression();
            op.Write(responses, wopts, this);
            ++index;
        }

        grpc::ServerAsyncWriter<EventList> op;
        EventHub::AsyncService &service;

        None request;
        EventList &responses;

        size_t index = 0;
        const grpc_compression_algorithm compressionAlgo;
        const size_t responseCount = 20;
    };

    auto processor = m_server->createProcessor();
    SubscribeListHandler *handler = new SubscribeListHandler(serverResponses, *m_service,
                                                             compressionAlgo, processor.get());
    m_server->startRpcTag(handler);

    auto call = m_client->SubscribeList(qt::None{});
    QVERIFY(call);

    connect(call.get(), &QGrpcOperation::finished, this,
            [&](const QGrpcStatus &status) { QCOMPARE(status.code(), QtGrpc::StatusCode::Ok); });
    connect(call.get(), &QGrpcServerStream::messageReceived, this, [&] {
        auto response = call->read<qt::EventList>();
        QVERIFY(response);
        QCOMPARE_EQ(response->events().size(), serverResponses.events().size());
        for (int i = 0; i < response->events().size(); ++i) {
            const auto &next = response->events().at(i);
            const auto &baseline = serverResponses.events().at(i);
            QCOMPARE_EQ(next.name(), QString::fromStdString(baseline.name()));
            QCOMPARE_EQ(next.number(), baseline.number());
        }
    });

    QSignalSpy finishedSpy(call.get(), &QGrpcOperation::finished);
    QVERIFY(finishedSpy.isValid());
    QVERIFY(finishedSpy.wait());
}

void QtGrpcClientEnd2EndTest::channelChangeCancelsInFlightRPCs_data() const
{
    QTest::addColumn<bool>("waitForServer");
    // Active: cancel hits an RPC the server has already accepted.
    QTest::addRow("Active") << true;
    // Idle: cancel races the first write; client still observes Cancelled.
    QTest::addRow("Idle") << false;
}

void QtGrpcClientEnd2EndTest::channelChangeCancelsInFlightRPCs()
{
    QFETCH(const bool, waitForServer);

    // Cancelled RPC leaves a half-drained server; rebuild for the next row.
    const auto restoreServer = qScopeGuard([this] { restartServer(); });

    auto processor = m_server->createProcessor();
    struct ServerData
    {
        grpc::ServerAsyncResponseWriter<None> op{ &ctx };
        grpc::ServerContext ctx;

        Event request;
        None response;
        std::atomic<bool> requestReceived = false;
        std::atomic<bool> notifyWhenDone = false;
    };
    auto data = std::make_unique<ServerData>();

    if (waitForServer) {
        data->ctx.AsyncNotifyWhenDone(new CallbackTag(
            [&](bool ok) {
                QVERIFY(ok);
                data->notifyWhenDone = true;
                return CallbackTag::Delete;
            },
            processor.get()));
        CallbackTag *callHandler = new CallbackTag(
            [&](bool ok) {
                QVERIFY(ok);
                data->requestReceived = true;
                return CallbackTag::Delete;
            },
            processor.get());
        m_service->RequestPush(&data->ctx, &data->request, &data->op, m_server->cq(),
                               m_server->cq(), callHandler);
    }

    auto call = m_client->Push(qt::Event{});
    QVERIFY(call);

    connect(call.get(), &QGrpcOperation::finished, this, [&](const QGrpcStatus &status) {
        QCOMPARE_EQ(status.code(), QtGrpc::StatusCode::Cancelled);
    });

    QSignalSpy finishedSpy(call.get(), &QGrpcOperation::finished);
    QSignalSpy channelChangedSpy(m_client.get(), &QGrpcClientBase::channelChanged);
    QVERIFY(finishedSpy.isValid());
    QVERIFY(channelChangedSpy.isValid());

    if (waitForServer) {
        // Ensure the server has committed to the RPC before we cancel.
        QTRY_VERIFY(data->requestReceived.load());
    }

    auto *channelPtr = static_cast<QGrpcHttp2Channel *>(m_client->channel().get());
    QVERIFY(channelPtr);
    const auto uri = channelPtr->hostUri();
    const auto opts = channelPtr->channelOptions();
    QVERIFY(m_client->attachChannel(std::make_shared<QGrpcHttp2Channel>(uri, opts)));

    // attachChannel() cancellation might emit finished() before we reach this line.
    QTRY_COMPARE_EQ(finishedSpy.count(), 1);
    QCOMPARE_EQ(channelChangedSpy.count(), 1);
    if (waitForServer)
        QTRY_VERIFY(data->notifyWhenDone.load());
}

void QtGrpcClientEnd2EndTest::maximumReceiveMessageSize_data() const
{
    constexpr std::optional<quint64> None = std::nullopt;
    constexpr std::optional<quint64> Small = 512;
    constexpr std::optional<quint64> Unlimited = 0;

    QTest::addColumn<std::optional<quint64>>("channelLimit");
    QTest::addColumn<std::optional<quint64>>("callLimit");
    QTest::addColumn<std::optional<quint64>>("envLimit");
    QTest::addColumn<StatusCode>("expectedCode");

    QTest::newRow("channel-option-rejects")
        << Small << None << None << StatusCode::ResourceExhausted;
    QTest::newRow("call-option-rejects") << None << Small << None << StatusCode::ResourceExhausted;
    QTest::newRow("call-unlimited-overrides-channel")
        << Small << Unlimited << None << StatusCode::Ok;
    QTest::newRow("env-var-rejects") << None << None << Small << StatusCode::ResourceExhausted;
    QTest::newRow("call-unlimited-beats-env") << None << Unlimited << Small << StatusCode::Ok;
    QTest::newRow("channel-unlimited-beats-env") << Unlimited << None << Small << StatusCode::Ok;
}

void QtGrpcClientEnd2EndTest::maximumReceiveMessageSize()
{
    QFETCH(const std::optional<quint64>, channelLimit);
    QFETCH(const std::optional<quint64>, callLimit);
    QFETCH(const std::optional<quint64>, envLimit);
    QFETCH(const QtGrpc::StatusCode, expectedCode);

    if (envLimit)
        qputenv("QT_GRPC_MAXIMUM_RECEIVE_MESSAGE_SIZE", QByteArray::number(*envLimit));
    const auto cleanup = qScopeGuard([] { qunsetenv("QT_GRPC_MAXIMUM_RECEIVE_MESSAGE_SIZE"); });

    // Re-attach a fresh channel so the env var (set above) is sampled at
    // channel construction time, not the one created in init().
    QFETCH_GLOBAL(const QUrl, hostUri);
    QFETCH_GLOBAL(const QGrpcChannelOptions, channelOptions);
    QGrpcChannelOptions chOpts = channelOptions;
    if (channelLimit)
        chOpts.setMaximumReceiveMessageSize(*channelLimit);
    QVERIFY(m_client->attachChannel(std::make_shared<QGrpcHttp2Channel>(hostUri, chOpts)));

    // Server sends a 2048-byte Event message, then finishes OK.
    auto processor = m_server->createProcessor();
    struct ServerData
    {
        grpc::ServerAsyncWriter<Event> op{ &ctx };
        grpc::ServerContext ctx;

        None request;
        Event response;
    };
    ServerData *data = new ServerData;
    data->response.set_name(std::string(2048, 'x'));

    CallbackTag *writeHandler = new CallbackTag(
        [&](bool ok) {
            QVERIFY(ok);
            data->op.Finish(grpc::Status::OK, new DeleteTag<ServerData>(data, processor.get()));
            return CallbackTag::Delete;
        },
        processor.get());
    CallbackTag *callHandler = new CallbackTag(
        [&](bool ok) {
            QVERIFY(ok);
            data->op.Write(data->response, writeHandler);
            return CallbackTag::Delete;
        },
        processor.get());
    m_service->RequestSubscribe(&data->ctx, &data->request, &data->op, m_server->cq(),
                                m_server->cq(), callHandler);

    QGrpcCallOptions copts;
    if (callLimit)
        copts.setMaximumReceiveMessageSize(*callLimit);
    auto call = m_client->Subscribe(qt::None{}, copts);
    QVERIFY(call);

    QSignalSpy finishedSpy(call.get(), &QGrpcOperation::finished);
    QVERIFY(finishedSpy.isValid());
    QVERIFY(finishedSpy.wait());

    const QVariant &finishedArg = finishedSpy.at(0).first();
    const auto *status = get_if<QGrpcStatus>(&finishedArg);
    QVERIFY(status);
    QCOMPARE_EQ(status->code(), expectedCode);
    if (status->code() == QtGrpc::StatusCode::ResourceExhausted)
        QVERIFY(status->message().contains("512"));
}

void QtGrpcClientEnd2EndTest::acceptedCompressionAlgorithms_data() const
{
    QTest::addColumn<CompressionAlgorithms>("setAlgorithms");
    QTest::addColumn<grpc_compression_algorithm>("serverAlgo");
    QTest::addColumn<StatusCode>("expectedCode");

    const CompressionAlgorithms defaults = QGrpcChannelOptions::supportedCompressionAlgorithms();
    const CompressionAlgorithms identity = CompressionAlgorithm::Identity;
    const CompressionAlgorithms gzip = CompressionAlgorithm::Gzip;
    const CompressionAlgorithms deflate = CompressionAlgorithm::Deflate;

    QTest::newRow("default-gzip-accepted") << defaults << GRPC_COMPRESS_GZIP << StatusCode::Ok;

    QTest::newRow("default-deflate-accepted")
        << defaults << GRPC_COMPRESS_DEFLATE << StatusCode::Ok;

    // Passing only Identity rejects every compressed encoding from the server,
    // while uncompressed responses are always accepted per the gRPC spec.
    QTest::newRow("identity-only-rejects-gzip")
        << identity << GRPC_COMPRESS_GZIP << StatusCode::Internal;

    QTest::newRow("identity-only-accepts-uncompressed")
        << identity << GRPC_COMPRESS_NONE << StatusCode::Ok;

    QTest::newRow("gzip-allowed-accepts-gzip") << gzip << GRPC_COMPRESS_GZIP << StatusCode::Ok;

    QTest::newRow("gzip-allowed-rejects-deflate")
        << gzip << GRPC_COMPRESS_DEFLATE << StatusCode::Internal;

    QTest::newRow("deflate-allowed-accepts-deflate")
        << deflate << GRPC_COMPRESS_DEFLATE << StatusCode::Ok;
}

void QtGrpcClientEnd2EndTest::acceptedCompressionAlgorithms()
{
    QFETCH(const CompressionAlgorithms, setAlgorithms);
    QFETCH(const grpc_compression_algorithm, serverAlgo);
    QFETCH(const StatusCode, expectedCode);

    QGrpcChannelOptions chOpts;
    chOpts.setAcceptedCompressionAlgorithms(setAlgorithms);
    m_client->channel()->setChannelOptions(chOpts);

    // Server: send one message with the requested compression algorithm, then finish OK.
    auto processor = m_server->createProcessor();
    auto *processorPtr = processor.get();
    struct ServerData
    {
        grpc::ServerAsyncWriter<Event> op{ &ctx };
        grpc::ServerContext ctx;
        None request;
        Event response;
    };
    ServerData *data = new ServerData;
    data->ctx.set_compression_algorithm(serverAlgo);
    data->response.set_name("hello");

    CallbackTag *writeHandler = new CallbackTag(
        [data, processorPtr](bool ok) {
            QVERIFY(ok);
            data->op.Finish(grpc::Status::OK, new DeleteTag<ServerData>(data, processorPtr));
            return CallbackTag::Delete;
        },
        processorPtr);
    CallbackTag *callHandler = new CallbackTag(
        [data, writeHandler](bool ok) {
            QVERIFY(ok);
            data->op.Write(data->response, writeHandler);
            return CallbackTag::Delete;
        },
        processorPtr);
    m_service->RequestSubscribe(&data->ctx, &data->request, &data->op, m_server->cq(),
                                m_server->cq(), callHandler);

    auto call = m_client->Subscribe(qt::None{});
    QVERIFY(call);

    QSignalSpy finishedSpy(call.get(), &QGrpcOperation::finished);
    QVERIFY(finishedSpy.isValid());
    QVERIFY(finishedSpy.wait());

    const auto status = qvariant_cast<QGrpcStatus>(finishedSpy.at(0).first());
    QCOMPARE_EQ(status.code(), expectedCode);
}

void QtGrpcClientEnd2EndTest::requestCompression_data() const
{
    using OptAlgo = std::optional<CompressionAlgorithm>;
    using C = CompressionAlgorithm;

    QTest::addColumn<OptAlgo>("channelAlgo");
    QTest::addColumn<OptAlgo>("callAlgo");
    QTest::addColumn<std::string>("payload");
    QTest::addColumn<bool>("serverRejects");
    QTest::addColumn<StatusCode>("expectedCode");

    const std::string big(1024, 'A');
    const std::string tiny("x");

    QTest::newRow("none") << OptAlgo{} << OptAlgo{} << big << false << StatusCode::Ok;
    QTest::newRow("channel-gzip") << OptAlgo{ C::Gzip } << OptAlgo{} << big << false
                                  << StatusCode::Ok;
    QTest::newRow("call-deflate") << OptAlgo{} << OptAlgo{ C::Deflate } << big << false
                                  << StatusCode::Ok;
    QTest::newRow("call-overrides-channel")
        << OptAlgo{ C::Gzip } << OptAlgo{ C::Deflate } << big << false << StatusCode::Ok;
    // Tiny payload would grow when compressed; the channel falls back to identity-encoded frames.
    QTest::newRow("no-shrink-fallback")
        << OptAlgo{ C::Gzip } << OptAlgo{} << tiny << false << StatusCode::Ok;
    // Server enabled only for identity rejects gzip-compressed requests at the
    // transport layer with Unimplemented; no application handler runs.
    QTest::newRow("server-rejects-gzip")
        << OptAlgo{ C::Gzip } << OptAlgo{} << big << true << StatusCode::Unimplemented;
}

void QtGrpcClientEnd2EndTest::requestCompression()
{
    QFETCH(const std::optional<CompressionAlgorithm>, channelAlgo);
    QFETCH(const std::optional<CompressionAlgorithm>, callAlgo);
    QFETCH(const std::string, payload);
    QFETCH(const bool, serverRejects);
    QFETCH(const StatusCode, expectedCode);

    // After a rejection scenario, restore the unrestricted server for the next row.
    const auto restoreServer = qScopeGuard([this, serverRejects] {
        if (serverRejects)
            restartServer();
    });

    if (serverRejects) {
        grpc::ChannelArguments args;
        args.SetInt(GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET, 1 << GRPC_COMPRESS_NONE);
        restartServer(args);
        // init()'s channel points at the now-dead server; re-attach to avoid
        // racing the reconnect (surfaces as Unavailable on unix transports).
        auto *staleChannel = static_cast<QGrpcHttp2Channel *>(m_client->channel().get());
        QVERIFY(staleChannel);
        auto freshChannel = std::make_shared<QGrpcHttp2Channel>(staleChannel->hostUri(),
                                                                staleChannel->channelOptions());
        QVERIFY(m_client->attachChannel(freshChannel));
    }

    QGrpcChannelOptions chOpts = m_client->channel()->channelOptions();
    if (channelAlgo)
        chOpts.setRequestCompression(*channelAlgo);
    m_client->channel()->setChannelOptions(chOpts);

    auto processor = m_server->createProcessor();
    auto *processorPtr = processor.get();
    struct ServerData
    {
        grpc::ServerAsyncResponseWriter<None> op{ &ctx };
        grpc::ServerContext ctx;
        Event request;
        None response;
    };

    // Only register a handler when the server accepts. In rejection paths gRPC
    // refuses before matching, leaving the tag pending.
    if (!serverRejects) {
        auto *data = new ServerData;
        auto *callHandler = new CallbackTag(
            [&payload, data, processorPtr](bool ok) {
                QVERIFY(ok);
                QCOMPARE_EQ(data->request.name(), payload);
                data->op.Finish(data->response, grpc::Status::OK,
                                new DeleteTag<ServerData>(data, processorPtr));
                return CallbackTag::Delete;
            },
            processorPtr);
        m_service->RequestPush(&data->ctx, &data->request, &data->op, m_server->cq(),
                               m_server->cq(), callHandler);
    }

    qt::Event evt;
    evt.setName(QString::fromStdString(payload));
    std::unique_ptr<QGrpcCallReply> call = callAlgo
        ? m_client->Push(evt, QGrpcCallOptions().setRequestCompression(*callAlgo))
        : m_client->Push(evt);
    QVERIFY(call);

    QSignalSpy finishedSpy(call.get(), &QGrpcOperation::finished);
    QVERIFY(finishedSpy.isValid());
    QVERIFY(finishedSpy.wait());
    QCOMPARE_EQ(qvariant_cast<QGrpcStatus>(finishedSpy.at(0).first()).code(), expectedCode);
}

QTEST_MAIN(QtGrpcClientEnd2EndTest)

#include "tst_grpc_client_end2end.moc"
