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

TEST_F(LowLevelE2E_Partition, CreatePartitionsSucceeds) {
    RecordProperty("description", "Creates partitions for an existing topic and verifies the resulting count.");
    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", "server_default",
                                         0, "server_default", {}));
    ASSERT_NO_THROW(
        client->create_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 43));

    ASSERT_NO_THROW({
        const auto stream_details = client->get_stream(make_string_identifier(stream_name));
        ASSERT_EQ(stream_details.topics.size(), 1u);
        EXPECT_EQ(stream_details.topics[0].name, topic_name);
        EXPECT_EQ(stream_details.topics[0].partitions_count, 44u);
    });
}

TEST_F(LowLevelE2E_Partition, CreatePartitionsBeforeLoginThrows) {
    RecordProperty("description",
                   "Throws when create_partitions is called before connect, and after connect but before login.");
    const std::string stream_name = GetRandomName();
    const std::string topic_name  = GetRandomName();

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

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

TEST_F(LowLevelE2E_Partition, CreatePartitionsOnNonExistentResourcesThrows) {
    RecordProperty("description", "Throws when create_partitions is called for a stream or topic that does not exist.");
    const std::string stream_name         = GetRandomName();
    const std::string topic_name          = GetRandomName();
    const std::string missing_stream_name = GetRandomName();
    const std::string missing_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", "server_default",
                                         0, "server_default", {}));

    ASSERT_THROW(
        client->create_partitions(make_string_identifier(missing_stream_name), make_string_identifier(topic_name), 1),
        std::exception);
    ASSERT_THROW(
        client->create_partitions(make_string_identifier(stream_name), make_string_identifier(missing_topic_name), 1),
        std::exception);
}

TEST_F(LowLevelE2E_Partition, CreatePartitionsWithInvalidIdentifiersThrows) {
    RecordProperty("description", "Rejects create_partitions requests that use invalid stream or topic identifiers.");
    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", "server_default",
                                         0, "server_default", {}));

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

    iggy::ffi::Identifier invalid_stream_numeric_id;
    invalid_stream_numeric_id.kind   = "numeric";
    invalid_stream_numeric_id.length = 1;
    invalid_stream_numeric_id.value.push_back(1);
    ASSERT_THROW(client->create_partitions(std::move(invalid_stream_numeric_id), make_string_identifier(topic_name), 1),
                 std::exception);

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

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

TEST_F(LowLevelE2E_Partition, CreatePartitionsWithBoundaryPartitionsCountValues) {
    RecordProperty("description",
                   "Accepts supported create_partitions counts and rejects values outside the allowed range.");
    const std::string stream_name = GetRandomName();

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

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

    struct TestCase {
        std::string topic_name;
        std::uint32_t partitions_count;
        bool should_succeed;
        std::uint32_t expected_total_partitions;
    };

    const std::vector<TestCase> test_cases = {
        {GetRandomName(), static_cast<std::uint32_t>(-1), false, 1},
        {GetRandomName(), 0, false, 1},
        {GetRandomName(), 1, true, 2},
        {GetRandomName(), 43, true, 44},
        {GetRandomName(), 1000, true, 1001},
        {GetRandomName(), 1001, false, 1},
    };

    for (const auto &test_case : test_cases) {
        SCOPED_TRACE(test_case.topic_name);
        ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), test_case.topic_name, 1, "none",
                                             "server_default", 0, "server_default", {}));

        if (test_case.should_succeed) {
            ASSERT_NO_THROW(client->create_partitions(make_string_identifier(stream_name),
                                                      make_string_identifier(test_case.topic_name),
                                                      test_case.partitions_count));
        } else {
            ASSERT_THROW(
                client->create_partitions(make_string_identifier(stream_name),
                                          make_string_identifier(test_case.topic_name), test_case.partitions_count),
                std::exception);
        }
    }

    ASSERT_NO_THROW({
        const auto stream_details = client->get_stream(make_string_identifier(stream_name));
        ASSERT_EQ(stream_details.topics.size(), test_cases.size());
        for (const auto &test_case : test_cases) {
            bool found = false;
            for (const auto &topic : stream_details.topics) {
                if (topic.name == test_case.topic_name) {
                    EXPECT_EQ(topic.partitions_count, test_case.expected_total_partitions);
                    found = true;
                    break;
                }
            }
            EXPECT_TRUE(found) << "Missing topic " << test_case.topic_name;
        }
    });
}

TEST_F(LowLevelE2E_Partition, CreatePartitionsWithNumericIdentifiersSucceeds) {
    RecordProperty("description",
                   "Creates partitions successfully when valid numeric stream and topic identifiers are used.");
    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", "server_default",
                                         0, "server_default", {}));

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

    ASSERT_NO_THROW(client->create_partitions(make_numeric_identifier(stream_details.id),
                                              make_numeric_identifier(stream_details.topics[0].id), 43));

    ASSERT_NO_THROW({
        const auto updated_stream_details = client->get_stream(make_numeric_identifier(stream_details.id));
        ASSERT_EQ(updated_stream_details.topics.size(), 1u);
        EXPECT_EQ(updated_stream_details.topics[0].id, stream_details.topics[0].id);
        EXPECT_EQ(updated_stream_details.topics[0].name, topic_name);
        EXPECT_EQ(updated_stream_details.topics[0].partitions_count, 44u);
    });
}

TEST_F(LowLevelE2E_Partition, DeletePartitionsSucceeds) {
    RecordProperty("description", "Deletes partitions from an existing topic and verifies the resulting count.");
    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, 44, "none", "server_default",
                                         0, "server_default", {}));
    ASSERT_NO_THROW(
        client->delete_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 43));

    ASSERT_NO_THROW({
        const auto stream_details = client->get_stream(make_string_identifier(stream_name));
        ASSERT_EQ(stream_details.topics.size(), 1u);
        EXPECT_EQ(stream_details.topics[0].name, topic_name);
        EXPECT_EQ(stream_details.topics[0].partitions_count, 1u);
    });
}

TEST_F(LowLevelE2E_Partition, DeleteMorePartitionsThanExistingThrows) {
    RecordProperty("description",
                   "Rejects delete_partitions counts outside the allowed range and counts greater than existing.");
    const std::string stream_name = GetRandomName();

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

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

    struct TestCase {
        std::string topic_name;
        std::uint32_t partitions_count;
        bool should_succeed;
        std::uint32_t initial_partitions;
        std::uint32_t expected_total_partitions;
    };

    const std::vector<TestCase> test_cases = {
        {GetRandomName(), static_cast<std::uint32_t>(-1), false, 3, 3},
        {GetRandomName(), 0, false, 3, 3},
        {GetRandomName(), 1, true, 3, 2},
        {GetRandomName(), 4, false, 3, 3},
    };

    for (const auto &test_case : test_cases) {
        SCOPED_TRACE(test_case.topic_name);
        ASSERT_NO_THROW(client->create_topic(make_string_identifier(stream_name), test_case.topic_name,
                                             test_case.initial_partitions, "none", "server_default", 0,
                                             "server_default", {}));

        if (test_case.should_succeed) {
            ASSERT_NO_THROW(client->delete_partitions(make_string_identifier(stream_name),
                                                      make_string_identifier(test_case.topic_name),
                                                      test_case.partitions_count));
        } else {
            ASSERT_THROW(
                client->delete_partitions(make_string_identifier(stream_name),
                                          make_string_identifier(test_case.topic_name), test_case.partitions_count),
                std::exception);
        }
    }

    ASSERT_NO_THROW({
        const auto stream_details = client->get_stream(make_string_identifier(stream_name));
        ASSERT_EQ(stream_details.topics.size(), test_cases.size());
        for (const auto &test_case : test_cases) {
            bool found = false;
            for (const auto &topic : stream_details.topics) {
                if (topic.name == test_case.topic_name) {
                    EXPECT_EQ(topic.partitions_count, test_case.expected_total_partitions);
                    found = true;
                    break;
                }
            }
            EXPECT_TRUE(found) << "Missing topic " << test_case.topic_name;
        }
    });
}

TEST_F(LowLevelE2E_Partition, DeletePartitionsBeforeCreatingAdditionalPartitionsSucceeds) {
    RecordProperty("description",
                   "Deletes partitions from the initial topic allocation without calling create_partitions first.");
    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, 3, "none", "server_default",
                                         0, "server_default", {}));
    ASSERT_NO_THROW(
        client->delete_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 1));

    ASSERT_NO_THROW({
        const auto stream_details = client->get_stream(make_string_identifier(stream_name));
        ASSERT_EQ(stream_details.topics.size(), 1u);
        EXPECT_EQ(stream_details.topics[0].partitions_count, 2u);
    });
}

TEST_F(LowLevelE2E_Partition, DeletePartitionsFromTopicWithZeroPartitionsThrows) {
    RecordProperty("description",
                   "Throws when delete_partitions is called with count 1 for a topic that currently has 0 partitions.");
    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, 0, "none", "server_default",
                                         0, "server_default", {}));

    ASSERT_THROW(client->delete_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 1),
                 std::exception);

    ASSERT_NO_THROW({
        const auto stream_details = client->get_stream(make_string_identifier(stream_name));
        ASSERT_EQ(stream_details.topics.size(), 1u);
        EXPECT_EQ(stream_details.topics[0].name, topic_name);
        EXPECT_EQ(stream_details.topics[0].partitions_count, 0u);
    });
}

TEST_F(LowLevelE2E_Partition, DeletePartitionsBeforeLoginThrows) {
    RecordProperty("description",
                   "Throws when delete_partitions is called before connect, and after connect but before login.");
    const std::string stream_name = GetRandomName();
    const std::string topic_name  = GetRandomName();

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

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

TEST_F(LowLevelE2E_Partition, DeletePartitionsOnNonExistentResourcesThrows) {
    RecordProperty("description", "Throws when delete_partitions is called for a stream or topic that does not exist.");
    const std::string stream_name         = GetRandomName();
    const std::string topic_name          = GetRandomName();
    const std::string missing_stream_name = GetRandomName();
    const std::string missing_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, 3, "none", "server_default",
                                         0, "server_default", {}));

    ASSERT_THROW(
        client->delete_partitions(make_string_identifier(missing_stream_name), make_string_identifier(topic_name), 1),
        std::exception);
    ASSERT_THROW(
        client->delete_partitions(make_string_identifier(stream_name), make_string_identifier(missing_topic_name), 1),
        std::exception);
}

TEST_F(LowLevelE2E_Partition, DeletePartitionsWithInvalidIdentifiersThrows) {
    RecordProperty("description", "Rejects delete_partitions requests that use invalid stream or topic identifiers.");
    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, 3, "none", "server_default",
                                         0, "server_default", {}));

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

    iggy::ffi::Identifier invalid_stream_numeric_id;
    invalid_stream_numeric_id.kind   = "numeric";
    invalid_stream_numeric_id.length = 1;
    invalid_stream_numeric_id.value.push_back(1);
    ASSERT_THROW(client->delete_partitions(std::move(invalid_stream_numeric_id), make_string_identifier(topic_name), 1),
                 std::exception);

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

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

TEST_F(LowLevelE2E_Partition, DeletePartitionsTwiceForSameTopicSucceeds) {
    RecordProperty("description", "Allows delete_partitions to be called twice for the same stream and topic.");
    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, 45, "none", "server_default",
                                         0, "server_default", {}));
    ASSERT_NO_THROW(
        client->delete_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 20));
    ASSERT_NO_THROW(
        client->delete_partitions(make_string_identifier(stream_name), make_string_identifier(topic_name), 20));

    ASSERT_NO_THROW({
        const auto stream_details = client->get_stream(make_string_identifier(stream_name));
        ASSERT_EQ(stream_details.topics.size(), 1u);
        EXPECT_EQ(stream_details.topics[0].name, topic_name);
        EXPECT_EQ(stream_details.topics[0].partitions_count, 5u);
    });
}

TEST_F(LowLevelE2E_Partition, DeletePartitionsAfterStreamDeletionThrows) {
    RecordProperty("description", "Throws when delete_partitions is called after the stream has been deleted.");
    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, 3, "none", "server_default",
                                         0, "server_default", {}));

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

    ASSERT_NO_THROW(client->delete_stream(make_numeric_identifier(stream_details.id)));
    ForgetTrackedStream(stream_name);

    ASSERT_THROW(client->delete_partitions(make_numeric_identifier(stream_details.id),
                                           make_numeric_identifier(stream_details.topics[0].id), 1),
                 std::exception);
}

// TODO(slbotbm): Add CreatePartitionsAfterTopicDeletionThrows test case after delete_topic function is added.