/*
 * 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 E2E_Stream : public E2ETestFixture {};

TEST_F(E2E_Stream, CreateStreamAfterLogin) {
    RecordProperty("description", "Creates a stream successfully after authenticating.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);
}

TEST_F(E2E_Stream, CreateDuplicateStreamThrows) {
    RecordProperty("description", "Rejects creating the same stream twice.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);
    ASSERT_THROW(client.CreateStream(stream_name), std::exception);
}

TEST_F(E2E_Stream, CreateStreamBeforeLoginThrows) {
    RecordProperty("description", "Throws when stream creation is attempted before authentication.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedOutHighLevelClient();

    ASSERT_THROW(client.CreateStream(stream_name), std::exception);
    ASSERT_NO_THROW(client.Connect());
    ASSERT_THROW(client.CreateStream(stream_name), std::exception);
    ASSERT_NO_THROW(client.Login("iggy", "iggy"));
    ASSERT_NO_THROW(client.Disconnect());
    ASSERT_THROW(client.CreateStream(stream_name), std::exception);
}

TEST_F(E2E_Stream, CreateStreamValidatesNameConstraintsAndUniqueness) {
    RecordProperty("description",
                   "Validates stream name length constraints and accepts the maximum allowed name length.");
    const std::string illegal_stream_names[] = {
        "",
        std::string(256, 'b'),
    };
    auto client = GetLoggedInHighLevelClient();
    for (const auto &stream_name : illegal_stream_names) {
        SCOPED_TRACE(stream_name);
        ASSERT_THROW(client.CreateStream(stream_name), std::exception);
    }

    const std::string max_length_name(255, 'a');
    ASSERT_NO_THROW(client.CreateStream(max_length_name));
    TrackStream(max_length_name);
}

TEST_F(E2E_Stream, CreateStreamWithEmojiName) {
    RecordProperty("description", "Creates a stream with a UTF-8 emoji name.");
    const std::string stream_name = "🚀🚀🚀🚀Apache Iggy🚀🚀🚀🚀";
    auto client                   = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW({
        const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
        EXPECT_EQ(stream_details.Name(), stream_name);
        EXPECT_EQ(stream_details.TopicsCount(), 0u);
        EXPECT_EQ(stream_details.Topics().size(), 0u);
    });
}

TEST_F(E2E_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();
    auto client                           = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);

    iggy::StreamDetails original_stream_details = client.GetStream(iggy::Identifier::String(stream_name));
    const std::uint32_t stream_id               = original_stream_details.Id();
    ForgetTrackedStream(stream_name);
    TrackStream(stream_id);

    ASSERT_NO_THROW(client.UpdateStream(iggy::Identifier::String(stream_name), updated_stream_name));

    ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)), std::exception);

    iggy::StreamDetails updated_stream_details = client.GetStream(iggy::Identifier::Numeric(stream_id));

    EXPECT_EQ(updated_stream_details.Id(), original_stream_details.Id());
    EXPECT_EQ(updated_stream_details.CreatedAt(), original_stream_details.CreatedAt());
    EXPECT_EQ(updated_stream_details.Name(), updated_stream_name);
    EXPECT_EQ(updated_stream_details.SizeBytes(), original_stream_details.SizeBytes());
    EXPECT_EQ(updated_stream_details.MessagesCount(), original_stream_details.MessagesCount());
    EXPECT_EQ(updated_stream_details.TopicsCount(), original_stream_details.TopicsCount());
    EXPECT_EQ(updated_stream_details.Topics().size(), original_stream_details.Topics().size());
}

TEST_F(E2E_Stream, UpdateStreamWithSameNameIsIdempotent) {
    RecordProperty("description",
                   "Calling update_stream with the current name succeeds without changing stream details.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);

    auto first_read = client.GetStream(iggy::Identifier::String(stream_name));
    ASSERT_NO_THROW(client.UpdateStream(iggy::Identifier::Numeric(first_read.Id()), stream_name));
    auto second_read = client.GetStream(iggy::Identifier::Numeric(first_read.Id()));

    EXPECT_EQ(second_read.Id(), first_read.Id());
    EXPECT_EQ(second_read.CreatedAt(), first_read.CreatedAt());
    EXPECT_EQ(second_read.Name(), first_read.Name());
    EXPECT_EQ(second_read.SizeBytes(), first_read.SizeBytes());
    EXPECT_EQ(second_read.MessagesCount(), first_read.MessagesCount());
    EXPECT_EQ(second_read.TopicsCount(), first_read.TopicsCount());
    EXPECT_EQ(second_read.Topics().size(), first_read.Topics().size());
}

TEST_F(E2E_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();
    auto client                           = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);

    auto unauthenticated_client = GetLoggedOutHighLevelClient();

    ASSERT_THROW(unauthenticated_client.UpdateStream(iggy::Identifier::String(stream_name), updated_stream_name),
                 std::exception);
    ASSERT_NO_THROW(unauthenticated_client.Connect());
    ASSERT_THROW(unauthenticated_client.UpdateStream(iggy::Identifier::String(stream_name), updated_stream_name),
                 std::exception);
    ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
    ASSERT_NO_THROW(unauthenticated_client.Disconnect());
    ASSERT_THROW(unauthenticated_client.UpdateStream(iggy::Identifier::String(stream_name), updated_stream_name),
                 std::exception);
}

TEST_F(E2E_Stream, UpdateStreamWithVariousUtf8Characters) {
    RecordProperty("description", "Updates a stream name with various UTF-8 values.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);

    std::uint32_t stream_id = 0;
    ASSERT_NO_THROW({
        const auto stream_details = client.GetStream(iggy::Identifier::String(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.UpdateStream(iggy::Identifier::Numeric(stream_id), updated_stream_name));
        ASSERT_NO_THROW({
            const auto stream_details = client.GetStream(iggy::Identifier::Numeric(stream_id));
            EXPECT_EQ(stream_details.Name(), updated_stream_name);
        });
    }
}

TEST_F(E2E_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();
    auto client                           = GetLoggedInHighLevelClient();
    ASSERT_THROW(client.UpdateStream(iggy::Identifier::String(stream_name), updated_stream_name), std::exception);
}

TEST_F(E2E_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();
    auto client                          = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(first_stream_name));
    TrackStream(first_stream_name);
    ASSERT_NO_THROW(client.CreateStream(second_stream_name));
    TrackStream(second_stream_name);

    ASSERT_THROW(client.UpdateStream(iggy::Identifier::String(first_stream_name), second_stream_name), std::exception);
}

TEST_F(E2E_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();
    auto client                           = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);

    ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
    ForgetTrackedStream(stream_name);

    ASSERT_THROW(client.UpdateStream(iggy::Identifier::String(stream_name), updated_stream_name), std::exception);
}

TEST_F(E2E_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();
    auto client                           = GetLoggedInHighLevelClient();
    iggy::ffi::Client *ffi_client         = GetLoggedInClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);

    std::uint32_t stream_id = 0;
    ASSERT_NO_THROW({
        const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
        stream_id                 = stream_details.Id();
    });
    ForgetTrackedStream(stream_name);
    TrackStream(stream_id);

    ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::Numeric(stream_id), topic_name,
                                       iggy::TopicCreateOptions()
                                           .SetPartitionsCount(2)
                                           .SetCompressionAlgorithm(iggy::CompressionAlgorithm::None())
                                           .SetMessageExpiry(iggy::Expiry::NeverExpire())));

    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(ffi_client->send_messages(make_numeric_identifier(stream_id), make_numeric_identifier(0),
                                              "partition_id", partition_id_bytes(0), std::move(messages)));

    auto stream_before_update = client.GetStream(iggy::Identifier::Numeric(stream_id));
    iggy::ffi::Stats stats_before_update{};
    ASSERT_NO_THROW({ stats_before_update = ffi_client->get_stats(); });

    ASSERT_NO_THROW(client.UpdateStream(iggy::Identifier::Numeric(stream_id), updated_stream_name));

    ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)), std::exception);
    auto stream_after_update = client.GetStream(iggy::Identifier::Numeric(stream_id));
    iggy::ffi::Stats stats_after_update{};
    ASSERT_NO_THROW({ stats_after_update = ffi_client->get_stats(); });

    EXPECT_EQ(stream_after_update.Id(), stream_before_update.Id());
    EXPECT_EQ(stream_after_update.CreatedAt(), stream_before_update.CreatedAt());
    EXPECT_EQ(stream_after_update.Name(), updated_stream_name);
    EXPECT_EQ(stream_after_update.SizeBytes(), stream_before_update.SizeBytes());
    EXPECT_EQ(stream_after_update.MessagesCount(), stream_before_update.MessagesCount());
    EXPECT_EQ(stream_after_update.TopicsCount(), stream_before_update.TopicsCount());
    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.CreatedAt(), before_topic.CreatedAt());
    EXPECT_EQ(after_topic.Name(), before_topic.Name());
    EXPECT_EQ(after_topic.SizeBytes(), before_topic.SizeBytes());
    EXPECT_EQ(after_topic.MessageExpiry(), before_topic.MessageExpiry());
    EXPECT_EQ(after_topic.CompressionAlgorithm(), before_topic.CompressionAlgorithm());
    EXPECT_EQ(after_topic.MaxTopicSize(), before_topic.MaxTopicSize());
    EXPECT_EQ(after_topic.MessagesCount(), before_topic.MessagesCount());
    EXPECT_EQ(after_topic.PartitionsCount(), before_topic.PartitionsCount());

    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(E2E_Stream, UpdateStreamValidatesNameBounds) {
    RecordProperty("description",
                   "Rejects invalid stream name lengths during update and accepts the maximum allowed name length.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);

    std::uint32_t stream_id = 0;
    ASSERT_NO_THROW({
        const auto stream_details = client.GetStream(iggy::Identifier::String(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.UpdateStream(iggy::Identifier::Numeric(stream_id), invalid_stream_name), std::exception);
    }
}

TEST_F(E2E_Stream, StreamCreatedAndDeletedSuccessfully) {
    RecordProperty("description", "Creates a stream and deletes it successfully by string identifier.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);

    ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
    ForgetTrackedStream(stream_name);
}

TEST_F(E2E_Stream, DeleteNotCreatedStreamThrows) {
    RecordProperty("description", "Throws when deleting a stream that does not exist.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)), std::exception);
}

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

    auto client = GetLoggedOutHighLevelClient();

    ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)), std::exception);

    ASSERT_NO_THROW(client.Connect());

    ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)), std::exception);
    ASSERT_NO_THROW(client.Login("iggy", "iggy"));
    ASSERT_NO_THROW(client.Disconnect());
    ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)), std::exception);
}

TEST_F(E2E_Stream, DeleteStreamTwiceThrows) {
    RecordProperty("description", "Throws when deleting the same stream a second time.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);

    ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
    ForgetTrackedStream(stream_name);
    ASSERT_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)), std::exception);
}

TEST_F(E2E_Stream, GetStreamByStringIdentifierReturnsStreamDetails) {
    RecordProperty("description", "Returns expected stream details when looked up by string identifier.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);

    ASSERT_NO_THROW({
        const auto stream_details = client.GetStream(iggy::Identifier::String(stream_name));
        EXPECT_EQ(stream_details.Name(), stream_name);
        EXPECT_EQ(stream_details.TopicsCount(), 0u);
        EXPECT_EQ(stream_details.Topics().size(), 0u);
        EXPECT_EQ(stream_details.MessagesCount(), 0u);
        EXPECT_EQ(stream_details.SizeBytes(), 0u);
    });
}

TEST_F(E2E_Stream, GetNonExistentStreamDetailsThrows) {
    RecordProperty("description", "Throws when requesting details for a stream that does not exist.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)), std::exception);
}

TEST_F(E2E_Stream, GetStreamDetailsBeforeLoginThrows) {
    RecordProperty("description", "Throws when stream details are requested before authentication.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedOutHighLevelClient();

    ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)), std::exception);
    ASSERT_NO_THROW(client.Connect());
    ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)), std::exception);
    ASSERT_NO_THROW(client.Login("iggy", "iggy"));
    ASSERT_NO_THROW(client.Disconnect());
    ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)), std::exception);
}

TEST_F(E2E_Stream, GetDeletedStreamDetailsThrows) {
    RecordProperty("description", "Throws when requesting details for a stream after it has been deleted.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW(client.GetStream(iggy::Identifier::String(stream_name)));
    ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
    ForgetTrackedStream(stream_name);
    ASSERT_THROW(client.GetStream(iggy::Identifier::String(stream_name)), std::exception);
}

TEST_F(E2E_Stream, GetStreamsReturnsEmptyAfterCleanup) {
    RecordProperty("description", "Verifies get_streams returns empty vector after cleaning up all streams.");
    auto client  = GetLoggedInHighLevelClient();
    auto streams = client.GetStreams();
    for (const auto &s : streams) {
        client.DeleteStream(iggy::Identifier::Numeric(s.Id()));
    }

    streams = client.GetStreams();
    ASSERT_EQ(streams.size(), 0u);
}

TEST_F(E2E_Stream, GetStreamsReturnsStreamAfterCreation) {
    RecordProperty("description", "Verifies created stream appears in get_streams result.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    client.CreateStream(stream_name);
    TrackStream(stream_name);
    auto streams = client.GetStreams();
    ASSERT_GE(streams.size(), 1u);

    bool found = false;
    for (const auto &s : streams) {
        if (s.Name() == stream_name) {
            found = true;
            EXPECT_GT(s.CreatedAt(), 0u);
            EXPECT_EQ(s.SizeBytes(), 0u);
            EXPECT_EQ(s.MessagesCount(), 0u);
            EXPECT_EQ(s.TopicsCount(), 0u);
            break;
        }
    }
    ASSERT_TRUE(found) << "Stream '" << stream_name << "' not found in get_streams result";
}

TEST_F(E2E_Stream, GetStreamsFieldsVerification) {
    RecordProperty("description",
                   "Verifies get_streams returns correct field values after creating stream with topic and messages.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    iggy::ffi::Client *ffi_client = GetLoggedInClient();
    client.CreateStream(stream_name);
    TrackStream(stream_name);
    auto stream                  = client.GetStream(iggy::Identifier::String(stream_name));
    const std::string topic_name = GetRandomName();
    client.CreateTopic(iggy::Identifier::Numeric(stream.Id()), topic_name,
                       iggy::TopicCreateOptions()
                           .SetPartitionsCount(1)
                           .SetCompressionAlgorithm(iggy::CompressionAlgorithm::None())
                           .SetMessageExpiry(iggy::Expiry::NeverExpire()));

    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));
    }
    ffi_client->send_messages(make_numeric_identifier(stream.Id()), make_numeric_identifier(0), "partition_id",
                              partition_id_bytes(0), std::move(messages));

    auto streams = client.GetStreams();
    ASSERT_GE(streams.size(), 1u);

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

TEST_F(E2E_Stream, GetStreamsBeforeLoginThrows) {
    RecordProperty("description", "Throws when get_streams is called before authentication.");
    auto client = GetLoggedOutHighLevelClient();

    ASSERT_THROW(client.GetStreams(), std::exception);
    ASSERT_NO_THROW(client.Connect());
    ASSERT_THROW(client.GetStreams(), std::exception);
    ASSERT_NO_THROW(client.Login("iggy", "iggy"));
    ASSERT_NO_THROW(client.Disconnect());
    ASSERT_THROW(client.GetStreams(), std::exception);
}

TEST_F(E2E_Stream, GetStreamsConsistentWithGetStream) {
    RecordProperty("description", "Verifies get_streams result is consistent with get_stream for the same stream.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    client.CreateStream(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.GetStreams();
    for (const auto &s : streams) {
        if (s.Name() == stream_name) {
            list_name         = s.Name();
            list_id           = s.Id();
            list_topics_count = s.TopicsCount();
            list_created_at   = s.CreatedAt();
            list_size_bytes   = s.SizeBytes();
            break;
        }
    }
    ASSERT_FALSE(list_name.empty()) << "Stream '" << stream_name << "' not found in get_streams result";

    auto single        = client.GetStream(iggy::Identifier::String(stream_name));
    auto single_name   = single.Name();
    auto single_topics = single.TopicsCount();

    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.CreatedAt());
    EXPECT_EQ(list_size_bytes, single.SizeBytes());
}

TEST_F(E2E_Stream, GetStreamsRepeatedCallsReturnSameResult) {
    RecordProperty("description", "Verifies repeated get_streams calls return consistent results.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    client.CreateStream(stream_name);
    TrackStream(stream_name);

    auto streams1 = client.GetStreams();
    auto streams2 = client.GetStreams();
    auto streams3 = client.GetStreams();

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

    auto contains_stream = [&](const std::vector<iggy::Stream> &vec) {
        for (const auto &s : vec) {
            if (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(E2E_Stream, PurgeStreamOnNonExistentStreamThrows) {
    RecordProperty("description", "Throws when purging a stream that does not exist.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    ASSERT_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)), std::exception);
}

TEST_F(E2E_Stream, PurgeStreamAfterStreamDeletionThrows) {
    RecordProperty("description", "Throws when purging a stream after it has been deleted.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW(client.DeleteStream(iggy::Identifier::String(stream_name)));
    ForgetTrackedStream(stream_name);

    ASSERT_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)), std::exception);
}

TEST_F(E2E_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();
    auto client                         = GetLoggedInHighLevelClient();
    iggy::ffi::Client *ffi_client       = GetLoggedInClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW(
        client.CreateTopic(iggy::Identifier::String(stream_name), first_topic_name,
                           iggy::TopicCreateOptions()
                               .SetPartitionsCount(2)
                               .SetCompressionAlgorithm(iggy::CompressionAlgorithm::Gzip())
                               .SetMessageExpiry(iggy::Expiry::Duration(1000))
                               .SetMaxTopicSize(iggy::MaxTopicSize::FromBytes(1024ULL * 1024ULL * 1024ULL))));
    ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), second_topic_name,
                                       iggy::TopicCreateOptions()
                                           .SetPartitionsCount(3)
                                           .SetCompressionAlgorithm(iggy::CompressionAlgorithm::None())
                                           .SetMessageExpiry(iggy::Expiry::NeverExpire())));

    const auto stream_before_purge = client.GetStream(iggy::Identifier::String(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(ffi_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.GetStream(iggy::Identifier::String(stream_name));
    EXPECT_GT(stream_with_messages.MessagesCount(), 0u);
    EXPECT_GT(stream_with_messages.SizeBytes(), 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::uint32_t partitions_count;
    };
    std::unordered_map<std::string, TopicMetadata> topics_before_purge;
    for (const auto &topic : stream_with_messages.Topics()) {
        topics_before_purge[topic.Name()] = {topic.Id(),
                                             topic.CreatedAt(),
                                             topic.Name(),
                                             topic.MessageExpiry(),
                                             topic.CompressionAlgorithm(),
                                             topic.MaxTopicSize(),
                                             topic.PartitionsCount()};
    }

    ASSERT_NO_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)));

    const auto stream_after_purge = client.GetStream(iggy::Identifier::String(stream_name));
    EXPECT_EQ(stream_after_purge.Id(), stream_with_messages.Id());
    EXPECT_EQ(stream_after_purge.CreatedAt(), stream_with_messages.CreatedAt());
    EXPECT_EQ(stream_after_purge.Name(), stream_with_messages.Name());
    EXPECT_EQ(stream_after_purge.TopicsCount(), stream_with_messages.TopicsCount());
    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 = 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.CreatedAt(), metadata.created_at);
        EXPECT_EQ(topic.Name(), metadata.name);
        EXPECT_EQ(topic.MessageExpiry(), metadata.message_expiry);
        EXPECT_EQ(topic.CompressionAlgorithm(), metadata.compression_algorithm);
        EXPECT_EQ(topic.MaxTopicSize(), metadata.max_topic_size);
        EXPECT_EQ(topic.PartitionsCount(), metadata.partitions_count);
    }
}

TEST_F(E2E_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();
    auto client                         = GetLoggedInHighLevelClient();
    iggy::ffi::Client *ffi_client       = GetLoggedInClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), first_topic_name,
                                       iggy::TopicCreateOptions().SetPartitionsCount(1)));
    ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), second_topic_name,
                                       iggy::TopicCreateOptions().SetPartitionsCount(1)));

    const auto created_stream = client.GetStream(iggy::Identifier::String(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 = 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(ffi_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(ffi_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.GetStream(iggy::Identifier::String(stream_name));
    EXPECT_EQ(stream_before_purge.TopicsCount(), 2u);
    EXPECT_EQ(stream_before_purge.MessagesCount(), 5u);
    EXPECT_GT(stream_before_purge.SizeBytes(), 0u);

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

    const auto streams_before_purge = client.GetStreams();
    bool found_stream_before_purge  = false;
    for (const auto &stream : streams_before_purge) {
        if (stream.Name() == stream_name) {
            found_stream_before_purge = true;
            EXPECT_EQ(stream.TopicsCount(), 2u);
            EXPECT_EQ(stream.MessagesCount(), 5u);
            EXPECT_GT(stream.SizeBytes(), 0u);
            break;
        }
    }
    ASSERT_TRUE(found_stream_before_purge);

    ASSERT_NO_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)));

    const auto stream_after_purge = client.GetStream(iggy::Identifier::String(stream_name));
    EXPECT_EQ(stream_after_purge.TopicsCount(), 2u);
    EXPECT_EQ(stream_after_purge.MessagesCount(), 0u);
    EXPECT_EQ(stream_after_purge.SizeBytes(), 0u);
    ASSERT_EQ(stream_after_purge.Topics().size(), 2u);
    for (const auto &topic : stream_after_purge.Topics()) {
        EXPECT_EQ(topic.MessagesCount(), 0u);
        EXPECT_EQ(topic.SizeBytes(), 0u);
    }

    const auto streams_after_purge = client.GetStreams();
    bool found_stream_after_purge  = false;
    for (const auto &stream : streams_after_purge) {
        if (stream.Name() == stream_name) {
            found_stream_after_purge = true;
            EXPECT_EQ(stream.TopicsCount(), 2u);
            EXPECT_EQ(stream.MessagesCount(), 0u);
            EXPECT_EQ(stream.SizeBytes(), 0u);
            break;
        }
    }
    ASSERT_TRUE(found_stream_after_purge);
}

TEST_F(E2E_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();
    auto client                         = GetLoggedInHighLevelClient();
    iggy::ffi::Client *ffi_client       = GetLoggedInClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), first_topic_name,
                                       iggy::TopicCreateOptions().SetPartitionsCount(2)));
    ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), second_topic_name,
                                       iggy::TopicCreateOptions().SetPartitionsCount(3)));

    const auto created_stream = client.GetStream(iggy::Identifier::String(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 = 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(ffi_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(ffi_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.GetStream(iggy::Identifier::String(stream_name));
    EXPECT_EQ(stream_before_purge.MessagesCount(), 10u);

    ASSERT_NO_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)));

    const auto stream_after_purge = client.GetStream(iggy::Identifier::String(stream_name));
    EXPECT_EQ(stream_after_purge.TopicsCount(), 2u);
    EXPECT_EQ(stream_after_purge.MessagesCount(), 0u);
    EXPECT_EQ(stream_after_purge.SizeBytes(), 0u);
    for (const auto &topic : stream_after_purge.Topics()) {
        EXPECT_EQ(topic.MessagesCount(), 0u);
        EXPECT_EQ(topic.SizeBytes(), 0u);
    }
}

TEST_F(E2E_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();
    auto client                   = GetLoggedInHighLevelClient();
    iggy::ffi::Client *ffi_client = GetLoggedInClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
                                       iggy::TopicCreateOptions().SetPartitionsCount(1)));

    const auto created_stream = client.GetStream(iggy::Identifier::String(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(ffi_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.PurgeStream(iggy::Identifier::String(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(ffi_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.GetStream(iggy::Identifier::String(stream_name));
    EXPECT_EQ(stream_after_resend.TopicsCount(), 1u);
    EXPECT_EQ(stream_after_resend.MessagesCount(), 2u);
    EXPECT_GT(stream_after_resend.SizeBytes(), 0u);
    ASSERT_EQ(stream_after_resend.Topics().size(), 1u);
    EXPECT_EQ(stream_after_resend.Topics().front().MessagesCount(), 2u);
}

TEST_F(E2E_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();
    auto client                   = GetLoggedInHighLevelClient();
    iggy::ffi::Client *ffi_client = GetLoggedInClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW(client.CreateTopic(iggy::Identifier::String(stream_name), topic_name,
                                       iggy::TopicCreateOptions().SetPartitionsCount(1)));

    const auto created_stream = client.GetStream(iggy::Identifier::String(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(ffi_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.PurgeStream(iggy::Identifier::String(stream_name)));
    const auto stream_after_first_purge = client.GetStream(iggy::Identifier::String(stream_name));
    EXPECT_EQ(stream_after_first_purge.TopicsCount(), 1u);
    EXPECT_EQ(stream_after_first_purge.MessagesCount(), 0u);
    EXPECT_EQ(stream_after_first_purge.SizeBytes(), 0u);
    ASSERT_EQ(stream_after_first_purge.Topics().size(), 1u);
    EXPECT_EQ(stream_after_first_purge.Topics().front().MessagesCount(), 0u);
    EXPECT_EQ(stream_after_first_purge.Topics().front().SizeBytes(), 0u);

    ASSERT_NO_THROW(client.PurgeStream(iggy::Identifier::String(stream_name)));
    const auto stream_after_second_purge = client.GetStream(iggy::Identifier::String(stream_name));
    EXPECT_EQ(stream_after_second_purge.TopicsCount(), 1u);
    EXPECT_EQ(stream_after_second_purge.MessagesCount(), 0u);
    EXPECT_EQ(stream_after_second_purge.SizeBytes(), 0u);
    ASSERT_EQ(stream_after_second_purge.Topics().size(), 1u);
    EXPECT_EQ(stream_after_second_purge.Topics().front().MessagesCount(), 0u);
    EXPECT_EQ(stream_after_second_purge.Topics().front().SizeBytes(), 0u);
}

TEST_F(E2E_Stream, PurgeStreamBeforeLoginThrows) {
    RecordProperty("description", "Throws when stream purge is attempted before authentication.");
    const std::string stream_name = GetRandomName();
    auto client                   = GetLoggedInHighLevelClient();
    ASSERT_NO_THROW(client.CreateStream(stream_name));
    TrackStream(stream_name);

    auto unauthenticated_client = GetLoggedOutHighLevelClient();

    ASSERT_THROW(unauthenticated_client.PurgeStream(iggy::Identifier::String(stream_name)), std::exception);
    ASSERT_NO_THROW(unauthenticated_client.Connect());
    ASSERT_THROW(unauthenticated_client.PurgeStream(iggy::Identifier::String(stream_name)), std::exception);
    ASSERT_NO_THROW(unauthenticated_client.Login("iggy", "iggy"));
    ASSERT_NO_THROW(unauthenticated_client.Disconnect());
    ASSERT_THROW(unauthenticated_client.PurgeStream(iggy::Identifier::String(stream_name)), std::exception);
}