/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

#include <cstdint>
#include <string>

#include <gtest/gtest.h>

#include "lib.rs.h"
#include "tests/e2e/test_helpers.hpp"

class LowLevelE2E_Stream : public E2ETestFixture {};

TEST_F(LowLevelE2E_Stream, CreateStreamAfterLogin) {
    RecordProperty("description", "Creates a stream successfully after authenticating.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();
    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);
}

TEST_F(LowLevelE2E_Stream, CreateDuplicateStreamThrows) {
    RecordProperty("description", "Rejects creating the same stream twice.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();
    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);
    ASSERT_THROW(client->create_stream(stream_name), std::exception);
}

TEST_F(LowLevelE2E_Stream, CreateStreamBeforeLoginThrows) {
    RecordProperty("description", "Throws when stream creation is attempted before authentication.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedOutClient();

    ASSERT_THROW(client->create_stream(stream_name), std::exception);
    ASSERT_NO_THROW(client->connect());
    ASSERT_THROW(client->create_stream(stream_name), std::exception);
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->disconnect());
    ASSERT_THROW(client->create_stream(stream_name), std::exception);
}

TEST_F(LowLevelE2E_Stream, CreateStreamValidatesNameConstraintsAndUniqueness) {
    RecordProperty("description",
                   "Validates stream name length constraints and accepts the maximum allowed name length.");
    iggy::ffi::Client *client = GetLoggedInClient();

    const std::string illegal_stream_names[] = {
        "",
        std::string(256, 'b'),
    };
    for (const auto &stream_name : illegal_stream_names) {
        SCOPED_TRACE(stream_name);
        ASSERT_THROW(client->create_stream(stream_name), std::exception);
    }

    const std::string max_length_name(255, 'a');
    ASSERT_NO_THROW(client->create_stream(max_length_name));
    TrackStream(max_length_name);
}

TEST_F(LowLevelE2E_Stream, CreateStreamWithEmojiName) {
    RecordProperty("description", "Creates a stream with a UTF-8 emoji name.");
    const std::string stream_name = "🚀🚀🚀🚀Apache Iggy🚀🚀🚀🚀";
    iggy::ffi::Client *client     = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW({
        const auto stream_details              = client->get_stream(make_string_identifier(stream_name));
        const std::string returned_stream_name = static_cast<std::string>(stream_details.name);
        EXPECT_EQ(returned_stream_name, stream_name);
        EXPECT_EQ(stream_details.topics_count, 0u);
        EXPECT_EQ(stream_details.topics.size(), 0u);
    });
}

TEST_F(LowLevelE2E_Stream, UpdateStreamWorksCorrectly) {
    RecordProperty("description", "Updates an existing stream name while preserving the stream identity.");
    const std::string stream_name         = GetRandomName();
    const std::string updated_stream_name = GetRandomName();
    iggy::ffi::Client *client             = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);

    iggy::ffi::StreamDetails original_stream_details{};
    ASSERT_NO_THROW({ original_stream_details = client->get_stream(make_string_identifier(stream_name)); });
    const std::uint32_t stream_id = original_stream_details.id;
    ForgetTrackedStream(stream_name);
    TrackStream(stream_id);

    ASSERT_NO_THROW(client->update_stream(make_string_identifier(stream_name), updated_stream_name));

    ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)), std::exception);

    iggy::ffi::StreamDetails updated_stream_details{};
    ASSERT_NO_THROW({ updated_stream_details = client->get_stream(make_numeric_identifier(stream_id)); });

    EXPECT_EQ(updated_stream_details.id, original_stream_details.id);
    EXPECT_EQ(updated_stream_details.created_at, original_stream_details.created_at);
    EXPECT_EQ(updated_stream_details.name, updated_stream_name);
    EXPECT_EQ(updated_stream_details.size_bytes, original_stream_details.size_bytes);
    EXPECT_EQ(updated_stream_details.messages_count, original_stream_details.messages_count);
    EXPECT_EQ(updated_stream_details.topics_count, original_stream_details.topics_count);
    EXPECT_EQ(updated_stream_details.topics.size(), original_stream_details.topics.size());
}

TEST_F(LowLevelE2E_Stream, UpdateStreamWithSameNameIsIdempotent) {
    RecordProperty("description",
                   "Calling update_stream with the current name succeeds without changing stream details.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);

    iggy::ffi::StreamDetails first_read{};
    iggy::ffi::StreamDetails second_read{};
    ASSERT_NO_THROW({ first_read = client->get_stream(make_string_identifier(stream_name)); });
    ASSERT_NO_THROW(client->update_stream(make_numeric_identifier(first_read.id), stream_name));
    ASSERT_NO_THROW({ second_read = client->get_stream(make_numeric_identifier(first_read.id)); });

    EXPECT_EQ(second_read.id, first_read.id);
    EXPECT_EQ(second_read.created_at, first_read.created_at);
    EXPECT_EQ(second_read.name, first_read.name);
    EXPECT_EQ(second_read.size_bytes, first_read.size_bytes);
    EXPECT_EQ(second_read.messages_count, first_read.messages_count);
    EXPECT_EQ(second_read.topics_count, first_read.topics_count);
    EXPECT_EQ(second_read.topics.size(), first_read.topics.size());
}

TEST_F(LowLevelE2E_Stream, UpdateStreamBeforeLoginThrows) {
    RecordProperty("description", "Rejects update_stream before connect, and after connect but before login.");
    const std::string stream_name         = GetRandomName();
    const std::string updated_stream_name = GetRandomName();
    iggy::ffi::Client *client             = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);

    iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();

    ASSERT_THROW(unauthenticated_client->update_stream(make_string_identifier(stream_name), updated_stream_name),
                 std::exception);
    ASSERT_NO_THROW(unauthenticated_client->connect());
    ASSERT_THROW(unauthenticated_client->update_stream(make_string_identifier(stream_name), updated_stream_name),
                 std::exception);
    ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(unauthenticated_client->disconnect());
    ASSERT_THROW(unauthenticated_client->update_stream(make_string_identifier(stream_name), updated_stream_name),
                 std::exception);
}

TEST_F(LowLevelE2E_Stream, UpdateStreamWithVariousUtf8Characters) {
    RecordProperty("description", "Updates a stream name with various UTF-8 values.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);

    std::uint32_t stream_id = 0;
    ASSERT_NO_THROW({
        const auto stream_details = client->get_stream(make_string_identifier(stream_name));
        stream_id                 = stream_details.id;
    });
    ForgetTrackedStream(stream_name);
    TrackStream(stream_id);

    const std::vector<std::string> updated_stream_names = {
        "こんにちは世界", "안녕하세요세계", "你好世界", "مرحبا بالعالم", "नमस्ते दुनिया", "🚀🍕✨🎯🔥",
    };

    for (const auto &updated_stream_name : updated_stream_names) {
        SCOPED_TRACE(updated_stream_name);
        ASSERT_NO_THROW(client->update_stream(make_numeric_identifier(stream_id), updated_stream_name));
        ASSERT_NO_THROW({
            const auto stream_details = client->get_stream(make_numeric_identifier(stream_id));
            EXPECT_EQ(stream_details.name, updated_stream_name);
        });
    }
}

TEST_F(LowLevelE2E_Stream, UpdateNonExistentStreamThrows) {
    RecordProperty("description", "Throws when updating a stream that does not exist.");
    const std::string stream_name         = GetRandomName();
    const std::string updated_stream_name = GetRandomName();
    iggy::ffi::Client *client             = GetLoggedInClient();

    ASSERT_THROW(client->update_stream(make_string_identifier(stream_name), updated_stream_name), std::exception);
}

TEST_F(LowLevelE2E_Stream, UpdateStreamWithDuplicateNameThrows) {
    RecordProperty("description", "Rejects renaming a stream to another stream's existing name.");
    const std::string first_stream_name  = GetRandomName();
    const std::string second_stream_name = GetRandomName();
    iggy::ffi::Client *client            = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(first_stream_name));
    TrackStream(first_stream_name);
    ASSERT_NO_THROW(client->create_stream(second_stream_name));
    TrackStream(second_stream_name);

    ASSERT_THROW(client->update_stream(make_string_identifier(first_stream_name), second_stream_name), std::exception);
}

TEST_F(LowLevelE2E_Stream, UpdateDeletedStreamThrows) {
    RecordProperty("description", "Throws when updating a stream after it has been deleted.");
    const std::string stream_name         = GetRandomName();
    const std::string updated_stream_name = GetRandomName();
    iggy::ffi::Client *client             = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);

    ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name)));
    ForgetTrackedStream(stream_name);

    ASSERT_THROW(client->update_stream(make_string_identifier(stream_name), updated_stream_name), std::exception);
}

TEST_F(LowLevelE2E_Stream, UpdateStreamFailedValidationDoesNotMutateStream) {
    RecordProperty("description", "Keeps the stream unchanged when update_stream fails wrapper validation.");
    const std::string stream_name         = GetRandomName();
    const std::string updated_stream_name = GetRandomName();
    iggy::ffi::Client *client             = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);

    iggy::ffi::StreamDetails stream_before_failed_update{};
    ASSERT_NO_THROW({ stream_before_failed_update = client->get_stream(make_string_identifier(stream_name)); });

    iggy::ffi::Identifier invalid_numeric_id;
    invalid_numeric_id.kind   = "numeric";
    invalid_numeric_id.length = 1;
    invalid_numeric_id.value.push_back(1);
    ASSERT_THROW(client->update_stream(std::move(invalid_numeric_id), updated_stream_name), std::exception);

    iggy::ffi::StreamDetails stream_after_failed_update{};
    ASSERT_NO_THROW({ stream_after_failed_update = client->get_stream(make_string_identifier(stream_name)); });

    EXPECT_EQ(stream_after_failed_update.id, stream_before_failed_update.id);
    EXPECT_EQ(stream_after_failed_update.created_at, stream_before_failed_update.created_at);
    EXPECT_EQ(stream_after_failed_update.name, stream_before_failed_update.name);
    EXPECT_EQ(stream_after_failed_update.size_bytes, stream_before_failed_update.size_bytes);
    EXPECT_EQ(stream_after_failed_update.messages_count, stream_before_failed_update.messages_count);
    EXPECT_EQ(stream_after_failed_update.topics_count, stream_before_failed_update.topics_count);
    EXPECT_EQ(stream_after_failed_update.topics.size(), stream_before_failed_update.topics.size());

    ASSERT_THROW(client->get_stream(make_string_identifier(updated_stream_name)), std::exception);
}

TEST_F(LowLevelE2E_Stream, UpdateStreamOnlyChangesName) {
    RecordProperty(
        "description",
        "Changes only the stream name and leaves stream, topic, message, partition, and segment data intact.");
    const std::string stream_name         = GetRandomName();
    const std::string updated_stream_name = GetRandomName();
    const std::string topic_name          = GetRandomName();
    iggy::ffi::Client *client             = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);

    std::uint32_t stream_id = 0;
    ASSERT_NO_THROW({
        const auto stream_details = client->get_stream(make_string_identifier(stream_name));
        stream_id                 = stream_details.id;
    });
    ForgetTrackedStream(stream_name);
    TrackStream(stream_id);

    ASSERT_NO_THROW(client->create_topic(make_numeric_identifier(stream_id), topic_name, 2, "none", 0, "never_expire",
                                         0, "server_default"));

    rust::Vec<iggy::ffi::IggyMessageToSend> messages;
    for (std::uint32_t i = 0; i < 3; ++i) {
        auto message = iggy::ffi::make_message(to_payload("stream-update-preserve-" + std::to_string(i)),
                                               rust::Vec<iggy::ffi::HeaderEntry>());
        messages.push_back(std::move(message));
    }
    ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream_id), make_numeric_identifier(0),
                                          "partition_id", partition_id_bytes(0), std::move(messages)));

    iggy::ffi::StreamDetails stream_before_update{};
    iggy::ffi::StreamDetails stream_after_update{};
    iggy::ffi::Stats stats_before_update{};
    iggy::ffi::Stats stats_after_update{};
    ASSERT_NO_THROW({
        stream_before_update = client->get_stream(make_numeric_identifier(stream_id));
        stats_before_update  = client->get_stats();
    });

    ASSERT_NO_THROW(client->update_stream(make_numeric_identifier(stream_id), updated_stream_name));

    ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)), std::exception);
    ASSERT_NO_THROW({
        stream_after_update = client->get_stream(make_numeric_identifier(stream_id));
        stats_after_update  = client->get_stats();
    });

    EXPECT_EQ(stream_after_update.id, stream_before_update.id);
    EXPECT_EQ(stream_after_update.created_at, stream_before_update.created_at);
    EXPECT_EQ(stream_after_update.name, updated_stream_name);
    EXPECT_EQ(stream_after_update.size_bytes, stream_before_update.size_bytes);
    EXPECT_EQ(stream_after_update.messages_count, stream_before_update.messages_count);
    EXPECT_EQ(stream_after_update.topics_count, stream_before_update.topics_count);
    ASSERT_EQ(stream_before_update.topics.size(), 1u);
    ASSERT_EQ(stream_after_update.topics.size(), 1u);

    const auto &before_topic = stream_before_update.topics[0];
    const auto &after_topic  = stream_after_update.topics[0];
    EXPECT_EQ(after_topic.id, before_topic.id);
    EXPECT_EQ(after_topic.created_at, before_topic.created_at);
    EXPECT_EQ(after_topic.name, before_topic.name);
    EXPECT_EQ(after_topic.size_bytes, before_topic.size_bytes);
    EXPECT_EQ(after_topic.message_expiry, before_topic.message_expiry);
    EXPECT_EQ(after_topic.compression_algorithm, before_topic.compression_algorithm);
    EXPECT_EQ(after_topic.max_topic_size, before_topic.max_topic_size);
    EXPECT_EQ(after_topic.replication_factor, before_topic.replication_factor);
    EXPECT_EQ(after_topic.messages_count, before_topic.messages_count);
    EXPECT_EQ(after_topic.partitions_count, before_topic.partitions_count);

    EXPECT_EQ(stats_after_update.streams_count, stats_before_update.streams_count);
    EXPECT_EQ(stats_after_update.topics_count, stats_before_update.topics_count);
    EXPECT_EQ(stats_after_update.partitions_count, stats_before_update.partitions_count);
    EXPECT_EQ(stats_after_update.segments_count, stats_before_update.segments_count);
    EXPECT_EQ(stats_after_update.messages_count, stats_before_update.messages_count);
}

TEST_F(LowLevelE2E_Stream, UpdateStreamValidatesNameBounds) {
    RecordProperty("description",
                   "Rejects invalid stream name lengths during update and accepts the maximum allowed name length.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);

    std::uint32_t stream_id = 0;
    ASSERT_NO_THROW({
        const auto stream_details = client->get_stream(make_string_identifier(stream_name));
        stream_id                 = stream_details.id;
    });
    ForgetTrackedStream(stream_name);
    TrackStream(stream_id);

    const std::vector<std::string> invalid_stream_names = {
        "",
        std::string(256, 'b'),
    };
    for (const auto &invalid_stream_name : invalid_stream_names) {
        SCOPED_TRACE("invalid_stream_name_length=" + std::to_string(invalid_stream_name.size()));
        ASSERT_THROW(client->update_stream(make_numeric_identifier(stream_id), invalid_stream_name), std::exception);
    }
}

TEST_F(LowLevelE2E_Stream, StreamCreatedAndDeletedSuccessfully) {
    RecordProperty("description", "Creates a stream and deletes it successfully by string identifier.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();
    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);

    ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name)));
    ForgetTrackedStream(stream_name);
}

TEST_F(LowLevelE2E_Stream, DeleteNotCreatedStreamThrows) {
    RecordProperty("description", "Throws when deleting a stream that does not exist.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();

    ASSERT_THROW(client->delete_stream(make_string_identifier(stream_name)), std::exception);
}

TEST_F(LowLevelE2E_Stream, DeleteStreamBeforeLoginThrows) {
    RecordProperty("description", "Throws when stream deletion is attempted before authentication.");
    const std::string stream_name = GetRandomName();

    iggy::ffi::Client *client = GetLoggedOutClient();

    ASSERT_THROW(client->delete_stream(make_string_identifier(stream_name)), std::exception);

    ASSERT_NO_THROW(client->connect());

    ASSERT_THROW(client->delete_stream(make_string_identifier(stream_name)), std::exception);
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->disconnect());
    ASSERT_THROW(client->delete_stream(make_string_identifier(stream_name)), std::exception);
}

TEST_F(LowLevelE2E_Stream, DeleteStreamTwiceThrows) {
    RecordProperty("description", "Throws when deleting the same stream a second time.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();
    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);

    ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name)));
    ForgetTrackedStream(stream_name);
    ASSERT_THROW(client->delete_stream(make_string_identifier(stream_name)), std::exception);
}

TEST_F(LowLevelE2E_Stream, DeleteStreamWithInvalidIdentifierThrows) {
    RecordProperty("description", "Rejects stream deletion requests that use invalid identifier formats.");
    iggy::ffi::Client *client = GetLoggedInClient();

    iggy::ffi::Identifier invalid_kind_id;
    invalid_kind_id.kind   = "invalid";
    invalid_kind_id.length = 4;
    invalid_kind_id.value  = {1, 0, 0, 0};
    ASSERT_THROW(client->delete_stream(std::move(invalid_kind_id)), std::exception);

    iggy::ffi::Identifier invalid_numeric_id;
    invalid_numeric_id.kind   = "numeric";
    invalid_numeric_id.length = 1;
    invalid_numeric_id.value.push_back(1);
    ASSERT_THROW(client->delete_stream(std::move(invalid_numeric_id)), std::exception);
}

TEST_F(LowLevelE2E_Stream, GetStreamDetailsWithInvalidIdentifierThrows) {
    RecordProperty("description", "Rejects stream detail lookups that use invalid identifier formats.");
    iggy::ffi::Client *client = GetLoggedInClient();

    iggy::ffi::Identifier invalid_kind_id;
    invalid_kind_id.kind   = "invalid";
    invalid_kind_id.length = 4;
    invalid_kind_id.value  = {1, 0, 0, 0};
    ASSERT_THROW(client->get_stream(std::move(invalid_kind_id)), std::exception);

    iggy::ffi::Identifier invalid_numeric_id;
    invalid_numeric_id.kind   = "numeric";
    invalid_numeric_id.length = 1;
    invalid_numeric_id.value.push_back(1);
    ASSERT_THROW(client->get_stream(std::move(invalid_numeric_id)), std::exception);
}

TEST_F(LowLevelE2E_Stream, GetStreamByStringIdentifierReturnsStreamDetails) {
    RecordProperty("description", "Returns expected stream details when looked up by string identifier.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();
    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);

    ASSERT_NO_THROW({
        const auto stream_details = client->get_stream(make_string_identifier(stream_name));
        EXPECT_EQ(stream_details.name, stream_name);
        EXPECT_EQ(stream_details.topics_count, 0u);
        EXPECT_EQ(stream_details.topics.size(), 0u);
        EXPECT_EQ(stream_details.messages_count, 0u);
        EXPECT_EQ(stream_details.size_bytes, 0u);
    });
}

TEST_F(LowLevelE2E_Stream, GetNonExistentStreamDetailsThrows) {
    RecordProperty("description", "Throws when requesting details for a stream that does not exist.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();
    ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)), std::exception);
}

TEST_F(LowLevelE2E_Stream, GetStreamDetailsBeforeLoginThrows) {
    RecordProperty("description", "Throws when stream details are requested before authentication.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedOutClient();

    ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)), std::exception);
    ASSERT_NO_THROW(client->connect());
    ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)), std::exception);
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->disconnect());
    ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)), std::exception);
}

TEST_F(LowLevelE2E_Stream, GetDeletedStreamDetailsThrows) {
    RecordProperty("description", "Throws when requesting details for a stream after it has been deleted.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();
    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW(client->get_stream(make_string_identifier(stream_name)));
    ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name)));
    ForgetTrackedStream(stream_name);
    ASSERT_THROW(client->get_stream(make_string_identifier(stream_name)), std::exception);
}

TEST_F(LowLevelE2E_Stream, GetStreamByNumericIdentifierReturnsStreamDetails) {
    RecordProperty("description", "Returns expected stream details when looked up by numeric identifier.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();
    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);

    std::uint32_t stream_id = 0;
    ASSERT_NO_THROW({
        const auto by_string = client->get_stream(make_string_identifier(stream_name));
        stream_id            = by_string.id;
    });

    ASSERT_NO_THROW({
        const auto by_numeric = client->get_stream(make_numeric_identifier(stream_id));
        EXPECT_EQ(by_numeric.id, stream_id);
        EXPECT_EQ(by_numeric.name, stream_name);
        EXPECT_EQ(by_numeric.topics_count, 0u);
        EXPECT_EQ(by_numeric.topics.size(), 0u);
    });
}

TEST_F(LowLevelE2E_Stream, GetStreamsReturnsEmptyAfterCleanup) {
    RecordProperty("description", "Verifies get_streams returns empty vector after cleaning up all streams.");
    iggy::ffi::Client *client = GetLoggedInClient();

    auto streams = client->get_streams();
    for (const auto &s : streams) {
        client->delete_stream(make_numeric_identifier(s.id));
    }

    streams = client->get_streams();
    ASSERT_EQ(streams.size(), 0u);
}

TEST_F(LowLevelE2E_Stream, GetStreamsReturnsStreamAfterCreation) {
    RecordProperty("description", "Verifies created stream appears in get_streams result.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();

    client->create_stream(stream_name);
    TrackStream(stream_name);
    auto streams = client->get_streams();
    ASSERT_GE(streams.size(), 1u);

    bool found = false;
    for (const auto &s : streams) {
        if (std::string(s.name) == stream_name) {
            found = true;
            EXPECT_GT(s.created_at, static_cast<std::uint64_t>(0));
            EXPECT_EQ(s.size_bytes, static_cast<std::uint64_t>(0));
            EXPECT_EQ(s.messages_count, static_cast<std::uint64_t>(0));
            EXPECT_EQ(s.topics_count, 0u);
            break;
        }
    }
    ASSERT_TRUE(found) << "Stream '" << stream_name << "' not found in get_streams result";
}

TEST_F(LowLevelE2E_Stream, GetStreamsFieldsVerification) {
    RecordProperty("description",
                   "Verifies get_streams returns correct field values after creating stream with topic and messages.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();

    client->create_stream(stream_name);
    TrackStream(stream_name);
    auto stream                  = client->get_stream(make_string_identifier(stream_name));
    const std::string topic_name = GetRandomName();
    client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", 0, "never_expire", 0,
                         "server_default");

    rust::Vec<iggy::ffi::IggyMessageToSend> messages;
    for (std::uint32_t i = 0; i < 5; i++) {
        auto msg = iggy::ffi::make_message(to_payload("field-verify-message-" + std::to_string(i)),
                                           rust::Vec<iggy::ffi::HeaderEntry>());
        messages.push_back(std::move(msg));
    }
    client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), "partition_id",
                          partition_id_bytes(0), std::move(messages));

    auto streams = client->get_streams();
    ASSERT_GE(streams.size(), 1u);

    bool found = false;
    for (const auto &s : streams) {
        if (std::string(s.name) == stream_name) {
            found = true;
            EXPECT_EQ(s.topics_count, 1u);
            EXPECT_EQ(s.messages_count, 5u);
            break;
        }
    }
    ASSERT_TRUE(found) << "Stream '" << stream_name << "' not found in get_streams result";
}

TEST_F(LowLevelE2E_Stream, GetStreamsBeforeLoginThrows) {
    RecordProperty("description", "Throws when get_streams is called before authentication.");
    iggy::ffi::Client *client = GetLoggedOutClient();

    ASSERT_THROW(client->get_streams(), std::exception);
    ASSERT_NO_THROW(client->connect());
    ASSERT_THROW(client->get_streams(), std::exception);
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->disconnect());
    ASSERT_THROW(client->get_streams(), std::exception);
}

TEST_F(LowLevelE2E_Stream, GetStreamsConsistentWithGetStream) {
    RecordProperty("description", "Verifies get_streams result is consistent with get_stream for the same stream.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();

    client->create_stream(stream_name);
    TrackStream(stream_name);

    std::string list_name;
    std::uint32_t list_id           = 0;
    std::uint32_t list_topics_count = 0;
    std::uint64_t list_created_at   = 0;
    std::uint64_t list_size_bytes   = 0;
    auto streams                    = client->get_streams();
    for (const auto &s : streams) {
        if (std::string(s.name) == stream_name) {
            list_name         = std::string(s.name);
            list_id           = s.id;
            list_topics_count = s.topics_count;
            list_created_at   = s.created_at;
            list_size_bytes   = s.size_bytes;
            break;
        }
    }
    ASSERT_FALSE(list_name.empty()) << "Stream '" << stream_name << "' not found in get_streams result";

    auto single        = client->get_stream(make_string_identifier(stream_name));
    auto single_name   = std::string(single.name);
    auto single_topics = single.topics_count;

    EXPECT_EQ(list_name, single_name);
    EXPECT_EQ(list_id, single.id);
    EXPECT_EQ(list_topics_count, single_topics);
    EXPECT_EQ(list_created_at, single.created_at);
    EXPECT_EQ(list_size_bytes, single.size_bytes);
}

TEST_F(LowLevelE2E_Stream, GetStreamsRepeatedCallsReturnSameResult) {
    RecordProperty("description", "Verifies repeated get_streams calls return consistent results.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();

    client->create_stream(stream_name);
    TrackStream(stream_name);

    auto streams1 = client->get_streams();
    auto streams2 = client->get_streams();
    auto streams3 = client->get_streams();

    ASSERT_EQ(streams1.size(), streams2.size());
    ASSERT_EQ(streams2.size(), streams3.size());

    auto contains_stream = [&](const rust::Vec<iggy::ffi::Stream> &vec) {
        for (const auto &s : vec) {
            if (std::string(s.name) == stream_name) {
                return true;
            }
        }
        return false;
    };

    ASSERT_TRUE(contains_stream(streams1)) << "Stream not found in first call";
    ASSERT_TRUE(contains_stream(streams2)) << "Stream not found in second call";
    ASSERT_TRUE(contains_stream(streams3)) << "Stream not found in third call";
}

TEST_F(LowLevelE2E_Stream, PurgeStreamOnNonExistentStreamThrows) {
    RecordProperty("description", "Throws when purging a stream that does not exist.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();

    ASSERT_THROW(client->purge_stream(make_string_identifier(stream_name)), std::exception);
}

TEST_F(LowLevelE2E_Stream, PurgeStreamAfterStreamDeletionThrows) {
    RecordProperty("description", "Throws when purging a stream after it has been deleted.");
    const std::string stream_name = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW(client->delete_stream(make_string_identifier(stream_name)));
    ForgetTrackedStream(stream_name);

    ASSERT_THROW(client->purge_stream(make_string_identifier(stream_name)), std::exception);
}

TEST_F(LowLevelE2E_Stream, PurgeStreamWithInvalidIdentifierThrows) {
    RecordProperty("description", "Rejects stream purge requests that use invalid identifier formats.");
    iggy::ffi::Client *client = GetLoggedInClient();

    iggy::ffi::Identifier invalid_kind_id;
    invalid_kind_id.kind   = "invalid";
    invalid_kind_id.length = 4;
    invalid_kind_id.value  = {1, 0, 0, 0};
    ASSERT_THROW(client->purge_stream(std::move(invalid_kind_id)), std::exception);

    iggy::ffi::Identifier invalid_numeric_id;
    invalid_numeric_id.kind   = "numeric";
    invalid_numeric_id.length = 1;
    invalid_numeric_id.value.push_back(1);
    ASSERT_THROW(client->purge_stream(std::move(invalid_numeric_id)), std::exception);
}

TEST_F(LowLevelE2E_Stream, PurgeStreamPreservesStreamMetadata) {
    RecordProperty("description", "Preserves stream identity and topic metadata after purging stream messages.");
    const std::string stream_name       = GetRandomName();
    const std::string first_topic_name  = GetRandomName();
    const std::string second_topic_name = GetRandomName();
    iggy::ffi::Client *client           = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), first_topic_name, 2, "gzip", 1,
                                         "duration", 1000, "1GiB"));
    ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), second_topic_name, 3, "none", 0,
                                         "never_expire", 0, "server_default"));

    const auto stream_before_purge = client->get_stream(make_string_identifier(stream_name));
    ASSERT_EQ(stream_before_purge.topics.size(), 2u);

    rust::Vec<iggy::ffi::IggyMessageToSend> first_topic_messages;
    first_topic_messages.push_back(
        iggy::ffi::make_message(to_payload("preserve-stream-metadata"), rust::Vec<iggy::ffi::HeaderEntry>()));
    ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream_before_purge.id),
                                          make_string_identifier(first_topic_name), "partition_id",
                                          partition_id_bytes(0), std::move(first_topic_messages)));

    const auto stream_with_messages = client->get_stream(make_string_identifier(stream_name));
    EXPECT_GT(stream_with_messages.messages_count, 0u);
    EXPECT_GT(stream_with_messages.size_bytes, 0u);

    struct TopicMetadata {
        std::uint32_t id;
        std::uint64_t created_at;
        std::string name;
        std::uint64_t message_expiry;
        std::string compression_algorithm;
        std::uint64_t max_topic_size;
        std::uint8_t replication_factor;
        std::uint32_t partitions_count;
    };
    std::unordered_map<std::string, TopicMetadata> topics_before_purge;
    for (const auto &topic : stream_with_messages.topics) {
        topics_before_purge[static_cast<std::string>(topic.name)] = {
            topic.id,
            topic.created_at,
            static_cast<std::string>(topic.name),
            topic.message_expiry,
            static_cast<std::string>(topic.compression_algorithm),
            topic.max_topic_size,
            topic.replication_factor,
            topic.partitions_count};
    }

    ASSERT_NO_THROW(client->purge_stream(make_string_identifier(stream_name)));

    const auto stream_after_purge = client->get_stream(make_string_identifier(stream_name));
    EXPECT_EQ(stream_after_purge.id, stream_with_messages.id);
    EXPECT_EQ(stream_after_purge.created_at, stream_with_messages.created_at);
    EXPECT_EQ(stream_after_purge.name, stream_with_messages.name);
    EXPECT_EQ(stream_after_purge.topics_count, stream_with_messages.topics_count);
    ASSERT_EQ(stream_after_purge.topics.size(), stream_with_messages.topics.size());

    for (const auto &topic : stream_after_purge.topics) {
        const std::string topic_name = static_cast<std::string>(topic.name);
        const auto metadata_it       = topics_before_purge.find(topic_name);
        ASSERT_NE(metadata_it, topics_before_purge.end());
        const auto &metadata = metadata_it->second;
        EXPECT_EQ(topic.id, metadata.id);
        EXPECT_EQ(topic.created_at, metadata.created_at);
        EXPECT_EQ(topic.name, metadata.name);
        EXPECT_EQ(topic.message_expiry, metadata.message_expiry);
        EXPECT_EQ(topic.compression_algorithm, metadata.compression_algorithm);
        EXPECT_EQ(topic.max_topic_size, metadata.max_topic_size);
        EXPECT_EQ(topic.replication_factor, metadata.replication_factor);
        EXPECT_EQ(topic.partitions_count, metadata.partitions_count);
    }
}

TEST_F(LowLevelE2E_Stream, PurgeStreamRemovesMessagesAndPreservesTopics) {
    RecordProperty("description", "Purges all stream messages while keeping the stream and topics intact.");
    const std::string stream_name       = GetRandomName();
    const std::string first_topic_name  = GetRandomName();
    const std::string second_topic_name = GetRandomName();
    iggy::ffi::Client *client           = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), first_topic_name, 1, "none", 0,
                                         "server_default", 0, "server_default"));
    ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), second_topic_name, 1, "none", 0,
                                         "server_default", 0, "server_default"));

    const auto created_stream = client->get_stream(make_string_identifier(stream_name));
    ASSERT_EQ(created_stream.topics.size(), 2u);

    std::uint32_t first_topic_id  = 0;
    std::uint32_t second_topic_id = 0;
    bool first_topic_found        = false;
    bool second_topic_found       = false;
    for (const auto &topic : created_stream.topics) {
        const std::string topic_name = static_cast<std::string>(topic.name);
        if (topic_name == first_topic_name) {
            first_topic_id    = topic.id;
            first_topic_found = true;
        } else if (topic_name == second_topic_name) {
            second_topic_id    = topic.id;
            second_topic_found = true;
        }
    }
    ASSERT_TRUE(first_topic_found);
    ASSERT_TRUE(second_topic_found);

    rust::Vec<iggy::ffi::IggyMessageToSend> first_topic_messages;
    for (std::uint32_t i = 0; i < 3; ++i) {
        first_topic_messages.push_back(iggy::ffi::make_message(to_payload("purge-stream-first-" + std::to_string(i)),
                                                               rust::Vec<iggy::ffi::HeaderEntry>()));
    }
    ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
                                          make_numeric_identifier(first_topic_id), "partition_id",
                                          partition_id_bytes(0), std::move(first_topic_messages)));

    rust::Vec<iggy::ffi::IggyMessageToSend> second_topic_messages;
    for (std::uint32_t i = 0; i < 2; ++i) {
        second_topic_messages.push_back(iggy::ffi::make_message(to_payload("purge-stream-second-" + std::to_string(i)),
                                                                rust::Vec<iggy::ffi::HeaderEntry>()));
    }
    ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
                                          make_numeric_identifier(second_topic_id), "partition_id",
                                          partition_id_bytes(0), std::move(second_topic_messages)));

    const auto stream_before_purge = client->get_stream(make_string_identifier(stream_name));
    EXPECT_EQ(stream_before_purge.topics_count, 2u);
    EXPECT_EQ(stream_before_purge.messages_count, 5u);
    EXPECT_GT(stream_before_purge.size_bytes, 0u);

    std::unordered_map<std::string, std::uint64_t> messages_before_purge;
    for (const auto &topic : stream_before_purge.topics) {
        messages_before_purge[static_cast<std::string>(topic.name)] = topic.messages_count;
    }
    EXPECT_EQ(messages_before_purge[first_topic_name], 3u);
    EXPECT_EQ(messages_before_purge[second_topic_name], 2u);

    const auto streams_before_purge = client->get_streams();
    bool found_stream_before_purge  = false;
    for (const auto &stream : streams_before_purge) {
        if (static_cast<std::string>(stream.name) == stream_name) {
            found_stream_before_purge = true;
            EXPECT_EQ(stream.topics_count, 2u);
            EXPECT_EQ(stream.messages_count, 5u);
            EXPECT_GT(stream.size_bytes, 0u);
            break;
        }
    }
    ASSERT_TRUE(found_stream_before_purge);

    ASSERT_NO_THROW(client->purge_stream(make_string_identifier(stream_name)));

    const auto stream_after_purge = client->get_stream(make_string_identifier(stream_name));
    EXPECT_EQ(stream_after_purge.topics_count, 2u);
    EXPECT_EQ(stream_after_purge.messages_count, 0u);
    EXPECT_EQ(stream_after_purge.size_bytes, 0u);
    ASSERT_EQ(stream_after_purge.topics.size(), 2u);
    for (const auto &topic : stream_after_purge.topics) {
        EXPECT_EQ(topic.messages_count, 0u);
        EXPECT_EQ(topic.size_bytes, 0u);
    }

    const auto streams_after_purge = client->get_streams();
    bool found_stream_after_purge  = false;
    for (const auto &stream : streams_after_purge) {
        if (static_cast<std::string>(stream.name) == stream_name) {
            found_stream_after_purge = true;
            EXPECT_EQ(stream.topics_count, 2u);
            EXPECT_EQ(stream.messages_count, 0u);
            EXPECT_EQ(stream.size_bytes, 0u);
            break;
        }
    }
    ASSERT_TRUE(found_stream_after_purge);
}

TEST_F(LowLevelE2E_Stream, PurgeStreamAcrossMultipleTopicsAndPartitionsClearsEverything) {
    RecordProperty("description", "Purges all messages across multiple topics and partitions in the stream.");
    const std::string stream_name       = GetRandomName();
    const std::string first_topic_name  = GetRandomName();
    const std::string second_topic_name = GetRandomName();
    iggy::ffi::Client *client           = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), first_topic_name, 2, "none", 0,
                                         "server_default", 0, "server_default"));
    ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), second_topic_name, 3, "none", 0,
                                         "server_default", 0, "server_default"));

    const auto created_stream = client->get_stream(make_string_identifier(stream_name));
    ASSERT_EQ(created_stream.topics.size(), 2u);

    std::uint32_t first_topic_id  = 0;
    std::uint32_t second_topic_id = 0;
    bool first_topic_found        = false;
    bool second_topic_found       = false;
    for (const auto &topic : created_stream.topics) {
        const std::string topic_name = static_cast<std::string>(topic.name);
        if (topic_name == first_topic_name) {
            first_topic_id    = topic.id;
            first_topic_found = true;
        } else if (topic_name == second_topic_name) {
            second_topic_id    = topic.id;
            second_topic_found = true;
        }
    }
    ASSERT_TRUE(first_topic_found);
    ASSERT_TRUE(second_topic_found);

    for (std::uint32_t partition_id = 0; partition_id < 2; ++partition_id) {
        rust::Vec<iggy::ffi::IggyMessageToSend> messages;
        for (std::uint32_t i = 0; i < 2; ++i) {
            messages.push_back(iggy::ffi::make_message(
                to_payload("purge-stream-topic-a-" + std::to_string(partition_id) + "-" + std::to_string(i)),
                rust::Vec<iggy::ffi::HeaderEntry>()));
        }
        ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
                                              make_numeric_identifier(first_topic_id), "partition_id",
                                              partition_id_bytes(partition_id), std::move(messages)));
    }
    for (std::uint32_t partition_id = 0; partition_id < 3; ++partition_id) {
        rust::Vec<iggy::ffi::IggyMessageToSend> messages;
        for (std::uint32_t i = 0; i < 2; ++i) {
            messages.push_back(iggy::ffi::make_message(
                to_payload("purge-stream-topic-b-" + std::to_string(partition_id) + "-" + std::to_string(i)),
                rust::Vec<iggy::ffi::HeaderEntry>()));
        }
        ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id),
                                              make_numeric_identifier(second_topic_id), "partition_id",
                                              partition_id_bytes(partition_id), std::move(messages)));
    }

    const auto stream_before_purge = client->get_stream(make_string_identifier(stream_name));
    EXPECT_EQ(stream_before_purge.messages_count, 10u);

    ASSERT_NO_THROW(client->purge_stream(make_string_identifier(stream_name)));

    const auto stream_after_purge = client->get_stream(make_string_identifier(stream_name));
    EXPECT_EQ(stream_after_purge.topics_count, 2u);
    EXPECT_EQ(stream_after_purge.messages_count, 0u);
    EXPECT_EQ(stream_after_purge.size_bytes, 0u);
    for (const auto &topic : stream_after_purge.topics) {
        EXPECT_EQ(topic.messages_count, 0u);
        EXPECT_EQ(topic.size_bytes, 0u);
    }
}

TEST_F(LowLevelE2E_Stream, PurgeStreamThenSendMessagesAgainSucceeds) {
    RecordProperty("description", "Allows sending fresh messages to a topic after purging its parent stream.");
    const std::string stream_name = GetRandomName();
    const std::string topic_name  = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", 0,
                                         "server_default", 0, "server_default"));

    const auto created_stream = client->get_stream(make_string_identifier(stream_name));
    ASSERT_EQ(created_stream.topics.size(), 1u);
    const std::uint32_t topic_id = created_stream.topics.front().id;

    rust::Vec<iggy::ffi::IggyMessageToSend> first_batch;
    first_batch.push_back(iggy::ffi::make_message(to_payload("before-purge"), rust::Vec<iggy::ffi::HeaderEntry>()));
    ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id), make_numeric_identifier(topic_id),
                                          "partition_id", partition_id_bytes(0), std::move(first_batch)));

    ASSERT_NO_THROW(client->purge_stream(make_string_identifier(stream_name)));

    rust::Vec<iggy::ffi::IggyMessageToSend> second_batch;
    second_batch.push_back(iggy::ffi::make_message(to_payload("after-purge-0"), rust::Vec<iggy::ffi::HeaderEntry>()));
    second_batch.push_back(iggy::ffi::make_message(to_payload("after-purge-1"), rust::Vec<iggy::ffi::HeaderEntry>()));
    ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id), make_numeric_identifier(topic_id),
                                          "partition_id", partition_id_bytes(0), std::move(second_batch)));

    const auto stream_after_resend = client->get_stream(make_string_identifier(stream_name));
    EXPECT_EQ(stream_after_resend.topics_count, 1u);
    EXPECT_EQ(stream_after_resend.messages_count, 2u);
    EXPECT_GT(stream_after_resend.size_bytes, 0u);
    ASSERT_EQ(stream_after_resend.topics.size(), 1u);
    EXPECT_EQ(stream_after_resend.topics.front().messages_count, 2u);
}

TEST_F(LowLevelE2E_Stream, PurgeStreamTwiceKeepsStreamEmptyAndTopicsIntact) {
    RecordProperty("description", "Allows purging the same stream twice and keeps the stream empty after both calls.");
    const std::string stream_name = GetRandomName();
    const std::string topic_name  = GetRandomName();
    iggy::ffi::Client *client     = GetLoggedInClient();

    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none", 0,
                                         "server_default", 0, "server_default"));

    const auto created_stream = client->get_stream(make_string_identifier(stream_name));
    ASSERT_EQ(created_stream.topics.size(), 1u);
    const std::uint32_t topic_id = created_stream.topics.front().id;

    rust::Vec<iggy::ffi::IggyMessageToSend> messages;
    for (std::uint32_t i = 0; i < 3; ++i) {
        messages.push_back(iggy::ffi::make_message(to_payload("purge-stream-twice-" + std::to_string(i)),
                                                   rust::Vec<iggy::ffi::HeaderEntry>()));
    }
    ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(created_stream.id), make_numeric_identifier(topic_id),
                                          "partition_id", partition_id_bytes(0), std::move(messages)));

    ASSERT_NO_THROW(client->purge_stream(make_string_identifier(stream_name)));
    const auto stream_after_first_purge = client->get_stream(make_string_identifier(stream_name));
    EXPECT_EQ(stream_after_first_purge.topics_count, 1u);
    EXPECT_EQ(stream_after_first_purge.messages_count, 0u);
    EXPECT_EQ(stream_after_first_purge.size_bytes, 0u);
    ASSERT_EQ(stream_after_first_purge.topics.size(), 1u);
    EXPECT_EQ(stream_after_first_purge.topics.front().messages_count, 0u);
    EXPECT_EQ(stream_after_first_purge.topics.front().size_bytes, 0u);

    ASSERT_NO_THROW(client->purge_stream(make_string_identifier(stream_name)));
    const auto stream_after_second_purge = client->get_stream(make_string_identifier(stream_name));
    EXPECT_EQ(stream_after_second_purge.topics_count, 1u);
    EXPECT_EQ(stream_after_second_purge.messages_count, 0u);
    EXPECT_EQ(stream_after_second_purge.size_bytes, 0u);
    ASSERT_EQ(stream_after_second_purge.topics.size(), 1u);
    EXPECT_EQ(stream_after_second_purge.topics.front().messages_count, 0u);
    EXPECT_EQ(stream_after_second_purge.topics.front().size_bytes, 0u);
}

TEST_F(LowLevelE2E_Stream, PurgeStreamBeforeLoginThrows) {
    RecordProperty("description", "Throws when stream purge is attempted before authentication.");
    const std::string stream_name = GetRandomName();

    iggy::ffi::Client *client = GetLoggedInClient();
    ASSERT_NO_THROW(client->create_stream(stream_name));
    TrackStream(stream_name);

    iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();

    ASSERT_THROW(unauthenticated_client->purge_stream(make_string_identifier(stream_name)), std::exception);
    ASSERT_NO_THROW(unauthenticated_client->connect());
    ASSERT_THROW(unauthenticated_client->purge_stream(make_string_identifier(stream_name)), std::exception);
    ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(unauthenticated_client->disconnect());
    ASSERT_THROW(unauthenticated_client->purge_stream(make_string_identifier(stream_name)), std::exception);
}