/*
 * 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.
 */

// TODO(slbotbm): Add tests for store_consumer_offset, get_consumer_offset, and delete_consumer_offset functions
// attached to client after implementing consumer group functions
// TODO(slbotbm): Add tests for update_permissions after creating create_user, get_user, etc. functions
#include <algorithm>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <limits>
#include <string>
#include <thread>
#include <unordered_set>

#include <gtest/gtest.h>

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

class LowLevelE2E_Client : public E2ETestFixture {};

TEST_F(LowLevelE2E_Client, ConnectAndLogin) {
    RecordProperty("description",
                   "Connects and returns matching login information using binary connection string formats.");
    constexpr std::uint32_t root_user_id   = 0;
    const std::string username             = "iggy";
    const std::string password             = "iggy";
    const std::string connection_strings[] = {
        "iggy://iggy:iggy@127.0.0.1:8090",
        "iggy+tcp://iggy:iggy@127.0.0.1:8090",
        "",
    };

    for (const std::string &connection_string : connection_strings) {
        SCOPED_TRACE(connection_string);
        iggy::ffi::Client *client = nullptr;
        ASSERT_NO_THROW({
            client = connection_string.empty() ? iggy::ffi::new_connection({})
                                               : iggy::ffi::from_connection_string(connection_string);
        });
        ASSERT_NE(client, nullptr);
        TrackClient(client);

        iggy::ffi::LoginInfo login_info{};
        iggy::ffi::ClientInfoDetails me{};
        ASSERT_NO_THROW(client->connect());
        ASSERT_NO_THROW({ login_info = client->login_user(username, password); });
        ASSERT_NO_THROW({ me = client->get_me(); });

        EXPECT_EQ(login_info.user_id, root_user_id);
        EXPECT_TRUE(me.has_user_id);
        EXPECT_EQ(login_info.user_id, me.user_id);
        EXPECT_FALSE(login_info.has_access_token);
        EXPECT_TRUE(static_cast<std::string>(login_info.access_token).empty());
        EXPECT_EQ(login_info.access_token_expiry, 0u);
    }
}

TEST_F(LowLevelE2E_Client, HttpLoginReturnsAccessToken) {
    RecordProperty("description", "Returns root user information and an access token after HTTP login.");
    constexpr std::uint32_t root_user_id = 0;
    iggy::ffi::Client *client            = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::from_connection_string("iggy+http://iggy:iggy@127.0.0.1:3000"); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

    iggy::ffi::LoginInfo login_info{};
    ASSERT_NO_THROW(client->connect());
    ASSERT_NO_THROW({ login_info = client->login_user("iggy", "iggy"); });

    EXPECT_EQ(login_info.user_id, root_user_id);
    EXPECT_TRUE(login_info.has_access_token);
    EXPECT_FALSE(static_cast<std::string>(login_info.access_token).empty());
    EXPECT_NE(login_info.access_token_expiry, 0u);
}

TEST_F(LowLevelE2E_Client, NewConnectionWithMalformedConnectionStringsThrow) {
    RecordProperty("description", "Rejects malformed connection strings when creating a new client connection.");
    const std::string malformed_connection_strings[] = {
        "iggy+invalid://iggy:iggy@127.0.0.1:8090", "iggy+tcp://iggy:iggy@:8090",      "iggy+tcp://iggy:iggy@127.0.0.1",
        "iggy+tcp://iggy:iggy@127.0.0.1:abc",      "iggy+tcp://:iggy@127.0.0.1:8090", "iggy+tcp://iggy:@127.0.0.1:8090",
        "iggy+tcp://iggy:iggy127.0.0.1:8090",      "not-a-connection-string",         "iggy://iggy:iggy@",
    };

    for (const std::string &connection_string : malformed_connection_strings) {
        SCOPED_TRACE(connection_string);
        ASSERT_THROW({ iggy::ffi::from_connection_string(connection_string); }, std::exception);
    }
}

TEST_F(LowLevelE2E_Client, LoginWithInvalidCredentialsThrows) {
    RecordProperty("description", "Throws when authentication uses invalid credentials after connecting.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

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

TEST_F(LowLevelE2E_Client, LoginTwiceWithDifferentCredentials) {
    RecordProperty("description", "Rejects a second login attempt that switches to invalid credentials.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

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

TEST_F(LowLevelE2E_Client, LogoutWithoutLogin) {
    RecordProperty("description",
                   "Rejects logout before authentication, both before and after connect, then succeeds after login.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

    ASSERT_THROW(client->logout_user(), std::exception);

    ASSERT_NO_THROW(client->connect());
    ASSERT_THROW(client->logout_user(), std::exception);

    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->logout_user());
}

TEST_F(LowLevelE2E_Client, ReloginOnSameClientAfterLogout) {
    RecordProperty("description", "Returns the same user identity when reauthenticating after a successful logout.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

    iggy::ffi::LoginInfo first_login{};
    iggy::ffi::LoginInfo second_login{};
    iggy::ffi::ClientInfoDetails first_me{};
    iggy::ffi::ClientInfoDetails second_me{};
    ASSERT_NO_THROW(client->connect());
    ASSERT_NO_THROW({ first_login = client->login_user("iggy", "iggy"); });
    ASSERT_NO_THROW({ first_me = client->get_me(); });
    ASSERT_NO_THROW(client->logout_user());
    ASSERT_NO_THROW({ second_login = client->login_user("iggy", "iggy"); });
    ASSERT_NO_THROW({ second_me = client->get_me(); });

    EXPECT_EQ(first_login.user_id, first_me.user_id);
    EXPECT_EQ(second_login.user_id, second_me.user_id);
    EXPECT_EQ(second_login.user_id, first_login.user_id);
    EXPECT_EQ(second_me.client_id, first_me.client_id);
    EXPECT_EQ(second_me.user_id, first_me.user_id);
}

TEST_F(LowLevelE2E_Client, LogoutErrorsWhenCalledMoreThanOnce) {
    RecordProperty("description",
                   "Rejects repeated logout calls once the authenticated session has already logged out.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

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

TEST_F(LowLevelE2E_Client, CreateUserWithUsernameOutsideLengthBoundsThrows) {
    RecordProperty("description", "Rejects 2-byte and 51-byte usernames over TCP without creating users.");
    iggy::ffi::Client *client = GetLoggedInClient();
    const std::string too_short_username(2, 'a');
    const std::string too_long_username(51, 'a');
    const std::string usernames[] = {too_short_username, too_long_username};

    ASSERT_EQ(too_short_username.size(), 2u);
    ASSERT_EQ(too_long_username.size(), 51u);
    for (const auto &username : usernames) {
        SCOPED_TRACE(username.size());
        ASSERT_THROW(
            client->create_user(username, "secret123", iggy::ffi::UserStatus::Active, false, iggy::ffi::Permissions{}),
            std::exception);
        ASSERT_THROW(client->get_user(make_string_identifier(username)), std::exception);
    }
}

TEST_F(LowLevelE2E_Client, CreateUserAcceptsNonAsciiAndNonAlphabeticUsernames) {
    RecordProperty("description",
                   "Creates and retrieves usernames containing punctuation, multilingual UTF-8, and emoji over TCP.");
    iggy::ffi::Client *client     = GetLoggedInClient();
    const std::string suffix      = GetRandomName(12);
    const std::string usernames[] = {
        "!@#_" + suffix, "ユーザー_" + suffix, "用户_" + suffix, "नाम_" + suffix, "사용자_" + suffix, "😀🚀_" + suffix,
    };

    for (const auto &username : usernames) {
        SCOPED_TRACE(username);
        ASSERT_LE(username.size(), 50u);

        iggy::ffi::UserInfoDetails created_user{};
        iggy::ffi::UserInfoDetails fetched_user{};
        ASSERT_NO_THROW({ created_user = CreateUser(client, username, "secret123", iggy::ffi::UserStatus::Active); });
        ASSERT_NO_THROW({ fetched_user = client->get_user(make_string_identifier(username)); });

        EXPECT_EQ(fetched_user.id, created_user.id);
        EXPECT_EQ(static_cast<std::string>(created_user.username), username);
        EXPECT_EQ(static_cast<std::string>(fetched_user.username), username);
    }
}

TEST_F(LowLevelE2E_Client, CreateUserBeforeLoginThrows) {
    RecordProperty("description", "Rejects user creation without an active authenticated session.");
    iggy::ffi::Client *client               = GetLoggedOutClient();
    iggy::ffi::Client *root                 = GetLoggedInClient();
    const std::string before_login_username = GetRandomName(50);
    const std::string logged_out_username   = GetRandomName(50);
    const std::string disconnected_username = GetRandomName(50);

    ASSERT_THROW(client->create_user(before_login_username, "secret123", iggy::ffi::UserStatus::Active, false,
                                     iggy::ffi::Permissions{}),
                 std::exception);
    ASSERT_NO_THROW(client->connect());
    ASSERT_THROW(client->create_user(before_login_username, "secret123", iggy::ffi::UserStatus::Active, false,
                                     iggy::ffi::Permissions{}),
                 std::exception);

    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->logout_user());
    ASSERT_THROW(client->create_user(logged_out_username, "secret123", iggy::ffi::UserStatus::Active, false,
                                     iggy::ffi::Permissions{}),
                 std::exception);

    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->disconnect());
    ASSERT_THROW(client->create_user(disconnected_username, "secret123", iggy::ffi::UserStatus::Active, false,
                                     iggy::ffi::Permissions{}),
                 std::exception);

    ASSERT_THROW(root->get_user(make_string_identifier(before_login_username)), std::exception);
    ASSERT_THROW(root->get_user(make_string_identifier(logged_out_username)), std::exception);
    ASSERT_THROW(root->get_user(make_string_identifier(disconnected_username)), std::exception);
}

TEST_F(LowLevelE2E_Client, CreateUserAcceptsUsernameAndPasswordLengthBounds) {
    RecordProperty("description",
                   "Creates users with shortest and longest ASCII usernames and passwords that can authenticate.");
    iggy::ffi::Client *root_client     = GetLoggedInClient();
    iggy::ffi::Client *shortest_client = GetLoggedOutClient();
    iggy::ffi::Client *longest_client  = GetLoggedOutClient();
    std::string shortest_username      = GetRandomName(3);
    std::string longest_username       = GetRandomName(50);
    const std::string shortest_password(3, 'a');
    const std::string longest_password(100, 'a');
    longest_username.resize(50, 'a');
    ASSERT_EQ(shortest_username.size(), 3u);
    ASSERT_EQ(longest_username.size(), 50u);
    ASSERT_EQ(shortest_password.size(), 3u);
    ASSERT_EQ(longest_password.size(), 100u);

    iggy::ffi::UserInfoDetails shortest_user{};
    iggy::ffi::UserInfoDetails longest_user{};
    iggy::ffi::UserInfoDetails fetched_shortest{};
    iggy::ffi::UserInfoDetails fetched_longest{};
    ASSERT_NO_THROW({
        shortest_user = CreateUser(root_client, shortest_username, shortest_password, iggy::ffi::UserStatus::Active);
    });
    ASSERT_NO_THROW(
        { longest_user = CreateUser(root_client, longest_username, longest_password, iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW({ fetched_shortest = root_client->get_user(make_string_identifier(shortest_username)); });
    ASSERT_NO_THROW({ fetched_longest = root_client->get_user(make_string_identifier(longest_username)); });
    ASSERT_NO_THROW(shortest_client->connect());
    ASSERT_NO_THROW(longest_client->connect());
    ASSERT_NO_THROW(shortest_client->login_user(shortest_username, shortest_password));
    ASSERT_NO_THROW(longest_client->login_user(longest_username, longest_password));

    EXPECT_EQ(static_cast<std::string>(shortest_user.username), shortest_username);
    EXPECT_EQ(static_cast<std::string>(longest_user.username), longest_username);
    EXPECT_EQ(fetched_shortest.id, shortest_user.id);
    EXPECT_EQ(fetched_longest.id, longest_user.id);
}

TEST_F(LowLevelE2E_Client, CreateUserWithPasswordOutsideLengthBoundsThrows) {
    RecordProperty("description", "Rejects 2-byte and 101-byte passwords without creating users.");
    iggy::ffi::Client *client        = GetLoggedInClient();
    const std::string short_username = GetRandomName(50);
    const std::string long_username  = GetRandomName(50);
    const std::string short_password(2, 'a');
    const std::string long_password(101, 'a');
    ASSERT_EQ(short_password.size(), 2u);
    ASSERT_EQ(long_password.size(), 101u);

    ASSERT_THROW(client->create_user(short_username, short_password, iggy::ffi::UserStatus::Active, false,
                                     iggy::ffi::Permissions{}),
                 std::exception);
    ASSERT_THROW(client->create_user(long_username, long_password, iggy::ffi::UserStatus::Active, false,
                                     iggy::ffi::Permissions{}),
                 std::exception);
    ASSERT_THROW(client->get_user(make_string_identifier(short_username)), std::exception);
    ASSERT_THROW(client->get_user(make_string_identifier(long_username)), std::exception);
}

TEST_F(LowLevelE2E_Client, CreateUserWithInvalidStatusThrows) {
    RecordProperty("description", "Rejects invalid status codes before creating users.");
    iggy::ffi::Client *client              = GetLoggedInClient();
    const iggy::ffi::UserStatus statuses[] = {
        static_cast<iggy::ffi::UserStatus>(0),
        static_cast<iggy::ffi::UserStatus>(3),
        static_cast<iggy::ffi::UserStatus>(std::numeric_limits<std::uint8_t>::max()),
    };

    for (const auto status : statuses) {
        const std::string username = GetRandomName(50);
        SCOPED_TRACE(static_cast<std::uint8_t>(status));
        ASSERT_THROW(client->create_user(username, "secret123", status, false, iggy::ffi::Permissions{}),
                     std::exception);
        ASSERT_THROW(client->get_user(make_string_identifier(username)), std::exception);
    }
}

TEST_F(LowLevelE2E_Client, CreateUserReturnsCreatedActiveUserDetails) {
    RecordProperty("description", "Returns and persists active user details.");
    iggy::ffi::Client *client  = GetLoggedInClient();
    const std::string username = GetRandomName(50);
    iggy::ffi::UserInfoDetails created_user{};
    iggy::ffi::UserInfoDetails fetched_user{};
    ASSERT_NO_THROW({ created_user = CreateUser(client, username, "secret123", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW({ fetched_user = client->get_user(make_string_identifier(username)); });

    EXPECT_EQ(fetched_user.id, created_user.id);
    EXPECT_EQ(static_cast<std::string>(created_user.username), username);
    EXPECT_EQ(static_cast<std::string>(fetched_user.username), username);
    EXPECT_EQ(created_user.status, iggy::ffi::UserStatus::Active);
    EXPECT_EQ(fetched_user.status, iggy::ffi::UserStatus::Active);
}

TEST_F(LowLevelE2E_Client, CreateUserRejectsDuplicateUsernameWithoutChangingOriginal) {
    RecordProperty("description", "Rejects duplicate usernames without changing the existing user.");
    iggy::ffi::Client *root_client = GetLoggedInClient();
    iggy::ffi::Client *user_client = GetLoggedOutClient();
    const std::string username     = GetRandomName(50);
    const std::string password     = "original-secret";
    iggy::ffi::UserInfoDetails original{};
    ASSERT_NO_THROW({ original = CreateUser(root_client, username, password, iggy::ffi::UserStatus::Active); });

    ASSERT_THROW(root_client->create_user(username, "replacement-secret", iggy::ffi::UserStatus::Inactive, false,
                                          iggy::ffi::Permissions{}),
                 std::exception);

    iggy::ffi::UserInfoDetails fetched{};
    ASSERT_NO_THROW({ fetched = root_client->get_user(make_string_identifier(username)); });
    EXPECT_EQ(fetched.id, original.id);
    EXPECT_EQ(fetched.status, iggy::ffi::UserStatus::Active);
    ASSERT_NO_THROW(user_client->connect());
    ASSERT_NO_THROW(user_client->login_user(username, password));
    iggy::ffi::Client *replacement_client = GetLoggedOutClient();
    ASSERT_NO_THROW(replacement_client->connect());
    ASSERT_THROW(replacement_client->login_user(username, "replacement-secret"), std::exception);
}

TEST_F(LowLevelE2E_Client, CreateUserPreservesNestedPermissionsInCreateAndGetResponses) {
    RecordProperty("description",
                   "Creates a user with global and per-resource permissions, then verifies create_user and get_user "
                   "return the same flags and numeric stream/topic IDs.");
    iggy::ffi::Client *client  = GetLoggedInClient();
    const std::string username = GetRandomName(50);
    iggy::ffi::Permissions permissions{};
    permissions.global.manage_servers = true;
    permissions.global.read_users     = true;
    permissions.global.manage_streams = true;
    permissions.global.read_topics    = true;
    permissions.global.send_messages  = true;

    iggy::ffi::StreamPermissionEntry first_stream{};
    first_stream.stream_id                 = 42;
    first_stream.permissions.manage_stream = true;
    first_stream.permissions.read_topics   = true;
    first_stream.permissions.send_messages = true;
    iggy::ffi::TopicPermissionEntry first_topic{};
    first_topic.topic_id                  = 7;
    first_topic.permissions.manage_topic  = true;
    first_topic.permissions.poll_messages = true;
    iggy::ffi::TopicPermissionEntry second_topic{};
    second_topic.topic_id                  = 9;
    second_topic.permissions.read_topic    = true;
    second_topic.permissions.send_messages = true;
    first_stream.permissions.topics.push_back(std::move(first_topic));
    first_stream.permissions.topics.push_back(std::move(second_topic));

    iggy::ffi::StreamPermissionEntry second_stream{};
    second_stream.stream_id                 = 84;
    second_stream.permissions.read_stream   = true;
    second_stream.permissions.manage_topics = true;
    second_stream.permissions.poll_messages = true;
    iggy::ffi::TopicPermissionEntry third_topic{};
    third_topic.topic_id               = 3;
    third_topic.permissions.read_topic = true;
    second_stream.permissions.topics.push_back(std::move(third_topic));
    permissions.streams.push_back(std::move(first_stream));
    permissions.streams.push_back(std::move(second_stream));

    iggy::ffi::UserInfoDetails created{};
    iggy::ffi::UserInfoDetails fetched{};
    ASSERT_NO_THROW({
        created =
            CreateUser(client, username, "secret123", iggy::ffi::UserStatus::Active, true, std::move(permissions));
    });
    ASSERT_NO_THROW({ fetched = client->get_user(make_string_identifier(username)); });
    for (const auto *user : {&created, &fetched}) {
        EXPECT_TRUE(user->permissions.global.manage_servers);
        EXPECT_FALSE(user->permissions.global.read_servers);
        EXPECT_FALSE(user->permissions.global.manage_users);
        EXPECT_TRUE(user->permissions.global.read_users);
        EXPECT_TRUE(user->permissions.global.manage_streams);
        EXPECT_FALSE(user->permissions.global.read_streams);
        EXPECT_FALSE(user->permissions.global.manage_topics);
        EXPECT_TRUE(user->permissions.global.read_topics);
        EXPECT_FALSE(user->permissions.global.poll_messages);
        EXPECT_TRUE(user->permissions.global.send_messages);
        ASSERT_EQ(user->permissions.streams.size(), 2u);

        const iggy::ffi::StreamPermissionEntry *stream_42 = nullptr;
        const iggy::ffi::StreamPermissionEntry *stream_84 = nullptr;
        for (const auto &stream : user->permissions.streams) {
            if (stream.stream_id == 42) {
                stream_42 = &stream;
            }
            if (stream.stream_id == 84) {
                stream_84 = &stream;
            }
        }
        ASSERT_NE(stream_42, nullptr);
        ASSERT_NE(stream_84, nullptr);
        EXPECT_TRUE(stream_42->permissions.manage_stream);
        EXPECT_FALSE(stream_42->permissions.read_stream);
        EXPECT_FALSE(stream_42->permissions.manage_topics);
        EXPECT_TRUE(stream_42->permissions.read_topics);
        EXPECT_FALSE(stream_42->permissions.poll_messages);
        EXPECT_TRUE(stream_42->permissions.send_messages);
        ASSERT_EQ(stream_42->permissions.topics.size(), 2u);
        const iggy::ffi::TopicPermissionEntry *topic_7 = nullptr;
        const iggy::ffi::TopicPermissionEntry *topic_9 = nullptr;
        for (const auto &topic : stream_42->permissions.topics) {
            if (topic.topic_id == 7) {
                topic_7 = &topic;
            }
            if (topic.topic_id == 9) {
                topic_9 = &topic;
            }
        }
        ASSERT_NE(topic_7, nullptr);
        ASSERT_NE(topic_9, nullptr);
        EXPECT_TRUE(topic_7->permissions.manage_topic);
        EXPECT_FALSE(topic_7->permissions.read_topic);
        EXPECT_TRUE(topic_7->permissions.poll_messages);
        EXPECT_FALSE(topic_7->permissions.send_messages);
        EXPECT_FALSE(topic_9->permissions.manage_topic);
        EXPECT_TRUE(topic_9->permissions.read_topic);
        EXPECT_FALSE(topic_9->permissions.poll_messages);
        EXPECT_TRUE(topic_9->permissions.send_messages);
        EXPECT_FALSE(stream_84->permissions.manage_stream);
        EXPECT_TRUE(stream_84->permissions.read_stream);
        EXPECT_TRUE(stream_84->permissions.manage_topics);
        EXPECT_FALSE(stream_84->permissions.read_topics);
        EXPECT_TRUE(stream_84->permissions.poll_messages);
        EXPECT_FALSE(stream_84->permissions.send_messages);
        ASSERT_EQ(stream_84->permissions.topics.size(), 1u);
        EXPECT_EQ(stream_84->permissions.topics[0].topic_id, 3u);
        EXPECT_FALSE(stream_84->permissions.topics[0].permissions.manage_topic);
        EXPECT_TRUE(stream_84->permissions.topics[0].permissions.read_topic);
        EXPECT_FALSE(stream_84->permissions.topics[0].permissions.poll_messages);
        EXPECT_FALSE(stream_84->permissions.topics[0].permissions.send_messages);
    }
}

TEST_F(LowLevelE2E_Client, CreatedUserCanReadOnlyTopicGrantedByPermissions) {
    RecordProperty("description",
                   "Creates a user with read access to one topic, then verifies that topic can be fetched and a topic "
                   "in another stream is denied.");
    iggy::ffi::Client *root_client        = GetLoggedInClient();
    iggy::ffi::Client *user_client        = GetLoggedOutClient();
    const std::string allowed_stream_name = GetRandomName();
    const std::string denied_stream_name  = GetRandomName();
    const std::string allowed_topic_name  = GetRandomName();
    const std::string denied_topic_name   = GetRandomName();
    const std::string username            = GetRandomName(50);

    iggy::ffi::StreamDetails allowed_stream{};
    iggy::ffi::StreamDetails denied_stream{};
    ASSERT_NO_THROW({ allowed_stream = root_client->create_stream(allowed_stream_name); });
    TrackStream(allowed_stream_name);
    ASSERT_NO_THROW({ denied_stream = root_client->create_stream(denied_stream_name); });
    TrackStream(denied_stream_name);

    iggy::ffi::TopicDetails allowed_topic{};
    iggy::ffi::TopicDetails denied_topic{};
    ASSERT_NO_THROW({
        allowed_topic = root_client->create_topic(make_numeric_identifier(allowed_stream.id), allowed_topic_name, 1,
                                                  "none", "server_default", 0, "server_default", {});
        denied_topic  = root_client->create_topic(make_numeric_identifier(denied_stream.id), denied_topic_name, 1,
                                                  "none", "server_default", 0, "server_default", {});
    });

    iggy::ffi::Permissions permissions{};
    iggy::ffi::StreamPermissionEntry stream_permissions{};
    stream_permissions.stream_id = allowed_stream.id;
    iggy::ffi::TopicPermissionEntry topic_permissions{};
    topic_permissions.topic_id               = allowed_topic.id;
    topic_permissions.permissions.read_topic = true;
    stream_permissions.permissions.topics.push_back(std::move(topic_permissions));
    permissions.streams.push_back(std::move(stream_permissions));
    ASSERT_NO_THROW({
        CreateUser(root_client, username, "secret123", iggy::ffi::UserStatus::Active, true, std::move(permissions));
    });
    ASSERT_NO_THROW(user_client->connect());
    ASSERT_NO_THROW(user_client->login_user(username, "secret123"));

    iggy::ffi::TopicDetails fetched_topic{};
    ASSERT_NO_THROW({
        fetched_topic = user_client->get_topic(make_numeric_identifier(allowed_stream.id),
                                               make_numeric_identifier(allowed_topic.id));
    });
    EXPECT_EQ(fetched_topic.id, allowed_topic.id);
    EXPECT_EQ(static_cast<std::string>(fetched_topic.name), allowed_topic_name);
    ASSERT_THROW(
        user_client->get_topic(make_numeric_identifier(denied_stream.id), make_numeric_identifier(denied_topic.id)),
        std::exception);
}

TEST_F(LowLevelE2E_Client, CreateUserRejectsDuplicateStreamPermissionIdsWithoutCreatingUser) {
    RecordProperty("description",
                   "Attempts to create a user with two permission entries for stream ID 42, then verifies creation "
                   "fails and no user is stored.");
    iggy::ffi::Client *client  = GetLoggedInClient();
    const std::string username = GetRandomName(50);
    iggy::ffi::Permissions permissions{};
    iggy::ffi::StreamPermissionEntry first_stream{};
    iggy::ffi::StreamPermissionEntry second_stream{};
    first_stream.stream_id  = 42;
    second_stream.stream_id = 42;
    permissions.streams.push_back(std::move(first_stream));
    permissions.streams.push_back(std::move(second_stream));

    ASSERT_THROW(
        client->create_user(username, "secret123", iggy::ffi::UserStatus::Active, true, std::move(permissions)),
        std::exception);
    ASSERT_THROW(client->get_user(make_string_identifier(username)), std::exception);
}

TEST_F(LowLevelE2E_Client, CreateUserRejectsDuplicateTopicPermissionIdsWithoutCreatingUser) {
    RecordProperty("description",
                   "Attempts to create a user with two permission entries for topic ID 7 in the same stream, then "
                   "verifies creation fails and no user is stored.");
    iggy::ffi::Client *client  = GetLoggedInClient();
    const std::string username = GetRandomName(50);
    iggy::ffi::Permissions permissions{};
    iggy::ffi::StreamPermissionEntry stream{};
    stream.stream_id = 42;
    iggy::ffi::TopicPermissionEntry first_topic{};
    iggy::ffi::TopicPermissionEntry second_topic{};
    first_topic.topic_id  = 7;
    second_topic.topic_id = 7;
    stream.permissions.topics.push_back(std::move(first_topic));
    stream.permissions.topics.push_back(std::move(second_topic));
    permissions.streams.push_back(std::move(stream));

    ASSERT_THROW(
        client->create_user(username, "secret123", iggy::ffi::UserStatus::Active, true, std::move(permissions)),
        std::exception);
    ASSERT_THROW(client->get_user(make_string_identifier(username)), std::exception);
}

TEST_F(LowLevelE2E_Client, ReadUsersPermissionDoesNotAllowCreateUser) {
    RecordProperty("description", "Rejects user creation by a user with read_users but not manage_users.");
    iggy::ffi::Client *root_client = GetLoggedInClient();
    iggy::ffi::Client *user_client = GetLoggedOutClient();
    const std::string username     = GetRandomName(50);
    const std::string target       = GetRandomName(50);
    iggy::ffi::Permissions permissions{};
    permissions.global.read_users = true;
    ASSERT_NO_THROW({
        CreateUser(root_client, username, "secret123", iggy::ffi::UserStatus::Active, true, std::move(permissions));
    });
    ASSERT_NO_THROW(user_client->connect());
    ASSERT_NO_THROW(user_client->login_user(username, "secret123"));

    ASSERT_THROW(
        user_client->create_user(target, "secret123", iggy::ffi::UserStatus::Active, false, iggy::ffi::Permissions{}),
        std::exception);
    ASSERT_THROW(root_client->get_user(make_string_identifier(target)), std::exception);
}

TEST_F(LowLevelE2E_Client, ManageUsersPermissionAllowsGrantingAdditionalPermissions) {
    RecordProperty("description", "Allows a user manager to grant a child a permission the manager does not have.");
    iggy::ffi::Client *root_client     = GetLoggedInClient();
    iggy::ffi::Client *manager_client  = GetLoggedOutClient();
    iggy::ffi::Client *child_client    = GetLoggedOutClient();
    const std::string manager_username = GetRandomName(50);
    const std::string child_username   = GetRandomName(50);
    const std::string denied_stream    = GetRandomName();
    const std::string child_stream     = GetRandomName();
    iggy::ffi::Permissions manager_permissions{};
    manager_permissions.global.manage_users   = true;
    manager_permissions.global.manage_streams = false;
    ASSERT_NO_THROW({
        CreateUser(root_client, manager_username, "secret123", iggy::ffi::UserStatus::Active, true,
                   std::move(manager_permissions));
    });
    ASSERT_NO_THROW(manager_client->connect());
    ASSERT_NO_THROW(manager_client->login_user(manager_username, "secret123"));
    ASSERT_THROW(manager_client->create_stream(denied_stream), std::exception);

    iggy::ffi::Permissions child_permissions{};
    child_permissions.global.manage_streams = true;
    iggy::ffi::UserInfoDetails child{};
    ASSERT_NO_THROW({
        child = CreateUser(manager_client, child_username, "child-secret", iggy::ffi::UserStatus::Active, true,
                           std::move(child_permissions));
    });
    EXPECT_EQ(static_cast<std::string>(child.username), child_username);
    EXPECT_EQ(child.status, iggy::ffi::UserStatus::Active);
    iggy::ffi::UserInfoDetails fetched{};
    ASSERT_NO_THROW({ fetched = root_client->get_user(make_string_identifier(child_username)); });
    EXPECT_EQ(fetched.id, child.id);
    EXPECT_TRUE(fetched.permissions.global.manage_streams);

    ASSERT_NO_THROW(child_client->connect());
    ASSERT_NO_THROW(child_client->login_user(child_username, "child-secret"));
    ASSERT_NO_THROW(child_client->create_stream(child_stream));
    TrackStream(child_stream);
}

TEST_F(LowLevelE2E_Client, CreatedActiveUserAuthenticatesOnlyWithSuppliedPassword) {
    RecordProperty("description", "Authenticates an active user only with its supplied password.");
    iggy::ffi::Client *root_client  = GetLoggedInClient();
    iggy::ffi::Client *valid_client = GetLoggedOutClient();
    iggy::ffi::Client *wrong_client = GetLoggedOutClient();
    const std::string username      = GetRandomName(50);
    const std::string password      = "known-secret";
    ASSERT_NO_THROW({ CreateUser(root_client, username, password, iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW(valid_client->connect());
    ASSERT_NO_THROW(wrong_client->connect());
    ASSERT_NO_THROW(valid_client->login_user(username, password));
    ASSERT_THROW(wrong_client->login_user(username, "other-secret"), std::exception);
}

TEST_F(LowLevelE2E_Client, CreatedInactiveUserCannotAuthenticate) {
    RecordProperty("description", "Persists inactive users but rejects authentication for them.");
    iggy::ffi::Client *root_client = GetLoggedInClient();
    iggy::ffi::Client *user_client = GetLoggedOutClient();
    const std::string username     = GetRandomName(50);
    const std::string password     = "inactive-secret";
    iggy::ffi::UserInfoDetails created{};
    iggy::ffi::UserInfoDetails fetched{};
    ASSERT_NO_THROW({ created = CreateUser(root_client, username, password, iggy::ffi::UserStatus::Inactive); });
    ASSERT_NO_THROW({ fetched = root_client->get_user(make_string_identifier(username)); });
    EXPECT_EQ(created.status, iggy::ffi::UserStatus::Inactive);
    EXPECT_EQ(fetched.status, iggy::ffi::UserStatus::Inactive);
    ASSERT_NO_THROW(user_client->connect());
    ASSERT_THROW(user_client->login_user(username, password), std::exception);
}

TEST_F(LowLevelE2E_Client, UpdateUserRejectsUnauthenticatedClientWithoutChangingTarget) {
    RecordProperty("description", "Rejects user updates without an active authenticated session.");
    iggy::ffi::Client *root_client = GetLoggedInClient();
    const std::string username     = GetRandomName(50);
    const std::string replacement  = GetRandomName(50);
    iggy::ffi::UserInfoDetails created{};
    ASSERT_NO_THROW({ created = CreateUser(root_client, username, "secret123", iggy::ffi::UserStatus::Active); });

    iggy::ffi::Client *client = GetLoggedOutClient();
    ASSERT_THROW(
        client->update_user(make_string_identifier(username), true, replacement, true, iggy::ffi::UserStatus::Inactive),
        std::exception);
    ASSERT_NO_THROW(client->connect());
    ASSERT_THROW(
        client->update_user(make_string_identifier(username), true, replacement, true, iggy::ffi::UserStatus::Inactive),
        std::exception);
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->logout_user());
    ASSERT_THROW(
        client->update_user(make_string_identifier(username), true, replacement, true, iggy::ffi::UserStatus::Inactive),
        std::exception);
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->disconnect());
    ASSERT_THROW(
        client->update_user(make_string_identifier(username), true, replacement, true, iggy::ffi::UserStatus::Inactive),
        std::exception);

    iggy::ffi::UserInfoDetails fetched{};
    ASSERT_NO_THROW({ fetched = root_client->get_user(make_string_identifier(username)); });
    EXPECT_EQ(fetched.id, created.id);
    EXPECT_EQ(static_cast<std::string>(fetched.username), username);
    EXPECT_EQ(fetched.status, iggy::ffi::UserStatus::Active);
}

TEST_F(LowLevelE2E_Client, UpdateUserRejectsUnknownUsernameAndNumericId) {
    RecordProperty("description", "Rejects updates for unknown username and numeric identifiers.");
    iggy::ffi::Client *client           = GetLoggedInClient();
    const std::string unknown_username  = GetRandomName(50);
    const std::string proposed_username = GetRandomName(50);
    const auto unknown_id               = std::numeric_limits<std::uint32_t>::max();

    ASSERT_THROW(client->update_user(make_string_identifier(unknown_username), true, proposed_username, true,
                                     iggy::ffi::UserStatus::Inactive),
                 std::exception);
    ASSERT_THROW(client->update_user(make_numeric_identifier(unknown_id), true, GetRandomName(50), true,
                                     iggy::ffi::UserStatus::Inactive),
                 std::exception);
    ASSERT_THROW(client->get_user(make_string_identifier(proposed_username)), std::exception);
}

TEST_F(LowLevelE2E_Client, UpdateUserByUsernameChangesUsernameAndStatus) {
    RecordProperty("description", "Updates a user by username and changes both username and status.");
    iggy::ffi::Client *client     = GetLoggedInClient();
    const std::string username    = GetRandomName(50);
    const std::string replacement = GetRandomName(50);
    iggy::ffi::UserInfoDetails created{};
    ASSERT_NO_THROW({ created = CreateUser(client, username, "secret123", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW(client->update_user(make_string_identifier(username), true, replacement, true,
                                        iggy::ffi::UserStatus::Inactive));
    RenameTrackedUser(username, replacement);

    ASSERT_THROW(client->get_user(make_string_identifier(username)), std::exception);
    iggy::ffi::UserInfoDetails fetched{};
    ASSERT_NO_THROW({ fetched = client->get_user(make_string_identifier(replacement)); });
    EXPECT_EQ(fetched.id, created.id);
    EXPECT_EQ(static_cast<std::string>(fetched.username), replacement);
    EXPECT_EQ(fetched.status, iggy::ffi::UserStatus::Inactive);
}

TEST_F(LowLevelE2E_Client, UpdateUserByNumericIdChangesUsernameAndStatus) {
    RecordProperty("description", "Updates a user by numeric ID and changes both username and status.");
    iggy::ffi::Client *client     = GetLoggedInClient();
    const std::string username    = GetRandomName(50);
    const std::string replacement = GetRandomName(50);
    iggy::ffi::UserInfoDetails created{};
    ASSERT_NO_THROW({ created = CreateUser(client, username, "secret123", iggy::ffi::UserStatus::Inactive); });
    ASSERT_NO_THROW(client->update_user(make_numeric_identifier(created.id), true, replacement, true,
                                        iggy::ffi::UserStatus::Active));
    RenameTrackedUser(username, replacement);

    iggy::ffi::UserInfoDetails by_id{};
    iggy::ffi::UserInfoDetails by_name{};
    ASSERT_NO_THROW({ by_id = client->get_user(make_numeric_identifier(created.id)); });
    ASSERT_NO_THROW({ by_name = client->get_user(make_string_identifier(replacement)); });
    EXPECT_EQ(by_id.id, created.id);
    EXPECT_EQ(by_name.id, created.id);
    EXPECT_EQ(static_cast<std::string>(by_id.username), replacement);
    EXPECT_EQ(static_cast<std::string>(by_name.username), replacement);
    EXPECT_EQ(by_id.status, iggy::ffi::UserStatus::Active);
    EXPECT_EQ(by_name.status, iggy::ffi::UserStatus::Active);
}

TEST_F(LowLevelE2E_Client, UpdateUserAllowsUsernameAndStatusToBeUpdatedIndependently) {
    RecordProperty("description", "Updates either username or status without changing the other field.");
    iggy::ffi::Client *client     = GetLoggedInClient();
    const std::string username    = GetRandomName(50);
    const std::string replacement = GetRandomName(50);
    iggy::ffi::UserInfoDetails created{};
    ASSERT_NO_THROW({ created = CreateUser(client, username, "secret123", iggy::ffi::UserStatus::Active); });

    ASSERT_NO_THROW(
        client->update_user(make_numeric_identifier(created.id), false, "", true, iggy::ffi::UserStatus::Inactive));
    iggy::ffi::UserInfoDetails status_updated{};
    ASSERT_NO_THROW({ status_updated = client->get_user(make_numeric_identifier(created.id)); });
    EXPECT_EQ(static_cast<std::string>(status_updated.username), username);
    EXPECT_EQ(status_updated.status, iggy::ffi::UserStatus::Inactive);

    ASSERT_NO_THROW(client->update_user(make_numeric_identifier(created.id), true, replacement, false,
                                        iggy::ffi::UserStatus::Active));
    RenameTrackedUser(username, replacement);

    iggy::ffi::UserInfoDetails username_updated{};
    ASSERT_NO_THROW({ username_updated = client->get_user(make_numeric_identifier(created.id)); });
    EXPECT_EQ(static_cast<std::string>(username_updated.username), replacement);
    EXPECT_EQ(username_updated.status, iggy::ffi::UserStatus::Inactive);
}

TEST_F(LowLevelE2E_Client, UpdateUserAcceptsUsernameLengthBounds) {
    RecordProperty("description", "Accepts exact three-byte and fifty-byte username boundaries.");
    iggy::ffi::Client *client           = GetLoggedInClient();
    const std::string first_username    = GetRandomName(50);
    const std::string second_username   = GetRandomName(50);
    const std::string first_replacement = GetRandomName(3);
    std::string second_replacement      = GetRandomName(50);
    second_replacement.resize(50, 'a');
    ASSERT_EQ(second_replacement.size(), 50u);
    iggy::ffi::UserInfoDetails first{};
    iggy::ffi::UserInfoDetails second{};
    ASSERT_NO_THROW({ first = CreateUser(client, first_username, "secret123", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW({ second = CreateUser(client, second_username, "secret123", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW(client->update_user(make_string_identifier(first_username), true, first_replacement, true,
                                        iggy::ffi::UserStatus::Active));
    RenameTrackedUser(first_username, first_replacement);
    ASSERT_NO_THROW(client->update_user(make_string_identifier(second_username), true, second_replacement, true,
                                        iggy::ffi::UserStatus::Active));
    RenameTrackedUser(second_username, second_replacement);

    iggy::ffi::UserInfoDetails fetched_first{};
    iggy::ffi::UserInfoDetails fetched_second{};
    ASSERT_NO_THROW({ fetched_first = client->get_user(make_string_identifier(first_replacement)); });
    ASSERT_NO_THROW({ fetched_second = client->get_user(make_string_identifier(second_replacement)); });
    EXPECT_EQ(fetched_first.id, first.id);
    EXPECT_EQ(fetched_second.id, second.id);
    EXPECT_EQ(static_cast<std::string>(fetched_first.username), first_replacement);
    EXPECT_EQ(static_cast<std::string>(fetched_second.username), second_replacement);
}

TEST_F(LowLevelE2E_Client, UpdateUserRejectsUsernameOutsideLengthBounds) {
    RecordProperty("description", "Rejects username sizes outside the SDK and server limits.");
    iggy::ffi::Client *client = GetLoggedInClient();
    const std::string source  = GetRandomName(50);
    iggy::ffi::UserInfoDetails created{};
    ASSERT_NO_THROW({ created = CreateUser(client, source, "secret123", iggy::ffi::UserStatus::Active); });
    const std::string invalid_usernames[] = {
        "", "a", "aa", std::string(51, 'c'), std::string(255, 'd'), std::string(256, 'e')};

    for (const auto &replacement : invalid_usernames) {
        ASSERT_THROW(
            client->update_user(make_string_identifier(source), true, replacement, true, iggy::ffi::UserStatus::Active),
            std::exception);
    }
    iggy::ffi::UserInfoDetails fetched{};
    ASSERT_NO_THROW({ fetched = client->get_user(make_string_identifier(source)); });
    EXPECT_EQ(fetched.id, created.id);
    EXPECT_EQ(fetched.status, iggy::ffi::UserStatus::Active);
    EXPECT_EQ(static_cast<std::string>(fetched.username), source);
}

TEST_F(LowLevelE2E_Client, UpdateUserAcceptsNonAsciiAndNonAlphabeticUsername) {
    RecordProperty("description", "Accepts representative non-ASCII and non-alphabetic usernames.");
    iggy::ffi::Client *client        = GetLoggedInClient();
    const std::string suffix         = GetRandomName(8);
    const std::string sources[]      = {GetRandomName(50), GetRandomName(50), GetRandomName(50)};
    const std::string replacements[] = {"!@#_" + suffix, "ユーザー_" + suffix, "😀🚀_" + suffix};
    ASSERT_LE(replacements[0].size(), 50u);
    ASSERT_LE(replacements[1].size(), 50u);
    ASSERT_LE(replacements[2].size(), 50u);
    for (std::size_t index = 0; index < 3; ++index) {
        iggy::ffi::UserInfoDetails created{};
        ASSERT_NO_THROW({ created = CreateUser(client, sources[index], "secret123", iggy::ffi::UserStatus::Active); });
        ASSERT_NO_THROW(client->update_user(make_string_identifier(sources[index]), true, replacements[index], true,
                                            iggy::ffi::UserStatus::Active));
        RenameTrackedUser(sources[index], replacements[index]);
        iggy::ffi::UserInfoDetails fetched{};
        ASSERT_NO_THROW({ fetched = client->get_user(make_string_identifier(replacements[index])); });
        EXPECT_EQ(fetched.id, created.id);
        EXPECT_EQ(static_cast<std::string>(fetched.username), replacements[index]);
    }
}

TEST_F(LowLevelE2E_Client, UpdateUserRejectsInvalidStatusWithoutRenamingTarget) {
    RecordProperty("description", "Rejects invalid status codes atomically with username changes.");
    iggy::ffi::Client *client  = GetLoggedInClient();
    const std::string username = GetRandomName(50);
    iggy::ffi::UserInfoDetails created{};
    ASSERT_NO_THROW({ created = CreateUser(client, username, "secret123", iggy::ffi::UserStatus::Active); });
    const iggy::ffi::UserStatus statuses[] = {
        static_cast<iggy::ffi::UserStatus>(0),
        static_cast<iggy::ffi::UserStatus>(3),
        static_cast<iggy::ffi::UserStatus>(std::numeric_limits<std::uint8_t>::max()),
    };
    for (const auto status : statuses) {
        const std::string replacement = GetRandomName(50);
        ASSERT_THROW(client->update_user(make_string_identifier(username), true, replacement, true, status),
                     std::exception);
        ASSERT_THROW(client->get_user(make_string_identifier(replacement)), std::exception);
    }
    iggy::ffi::UserInfoDetails fetched{};
    ASSERT_NO_THROW({ fetched = client->get_user(make_string_identifier(username)); });
    EXPECT_EQ(fetched.id, created.id);
    EXPECT_EQ(fetched.status, iggy::ffi::UserStatus::Active);
    EXPECT_EQ(static_cast<std::string>(fetched.username), username);
}

TEST_F(LowLevelE2E_Client, UpdateUserRejectsDuplicateUsernameWithoutChangingStatus) {
    RecordProperty("description", "Rejects duplicate usernames without partially applying the status update.");
    iggy::ffi::Client *client           = GetLoggedInClient();
    const std::string target_username   = GetRandomName(50);
    const std::string conflict_username = GetRandomName(50);
    iggy::ffi::UserInfoDetails target{};
    iggy::ffi::UserInfoDetails conflict{};
    ASSERT_NO_THROW({ target = CreateUser(client, target_username, "secret123", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW(
        { conflict = CreateUser(client, conflict_username, "secret123", iggy::ffi::UserStatus::Inactive); });
    ASSERT_THROW(client->update_user(make_string_identifier(target_username), true, conflict_username, true,
                                     iggy::ffi::UserStatus::Inactive),
                 std::exception);
    iggy::ffi::UserInfoDetails fetched_target{};
    iggy::ffi::UserInfoDetails fetched_conflict{};
    ASSERT_NO_THROW({ fetched_target = client->get_user(make_string_identifier(target_username)); });
    ASSERT_NO_THROW({ fetched_conflict = client->get_user(make_string_identifier(conflict_username)); });
    EXPECT_EQ(fetched_target.id, target.id);
    EXPECT_EQ(fetched_conflict.id, conflict.id);
    EXPECT_EQ(fetched_target.status, iggy::ffi::UserStatus::Active);
    EXPECT_EQ(fetched_conflict.status, iggy::ffi::UserStatus::Inactive);
}

TEST_F(LowLevelE2E_Client, UpdateUserAllowsCurrentUsernameWhileChangingStatus) {
    RecordProperty("description", "Allows a same-username update that changes status.");
    iggy::ffi::Client *client  = GetLoggedInClient();
    const std::string username = GetRandomName(50);
    iggy::ffi::UserInfoDetails created{};
    ASSERT_NO_THROW({ created = CreateUser(client, username, "secret123", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW(
        client->update_user(make_string_identifier(username), true, username, true, iggy::ffi::UserStatus::Inactive));
    iggy::ffi::UserInfoDetails fetched{};
    ASSERT_NO_THROW({ fetched = client->get_user(make_string_identifier(username)); });
    EXPECT_EQ(fetched.id, created.id);
    EXPECT_EQ(fetched.status, iggy::ffi::UserStatus::Inactive);
    EXPECT_EQ(static_cast<std::string>(fetched.username), username);
}

TEST_F(LowLevelE2E_Client, UpdateUserPreservesPasswordPermissionsAndCreationData) {
    RecordProperty("description", "Preserves password, permissions, ID, and creation timestamp after a rename.");
    iggy::ffi::Client *root_client  = GetLoggedInClient();
    iggy::ffi::Client *valid_client = GetLoggedOutClient();
    iggy::ffi::Client *old_client   = GetLoggedOutClient();
    const std::string username      = GetRandomName(50);
    const std::string replacement   = GetRandomName(50);
    iggy::ffi::Permissions permissions{};
    permissions.global.read_users    = true;
    permissions.global.read_streams  = true;
    permissions.global.send_messages = true;
    iggy::ffi::UserInfoDetails created{};
    ASSERT_NO_THROW({
        created = CreateUser(root_client, username, "known-secret", iggy::ffi::UserStatus::Active, true,
                             std::move(permissions));
    });
    ASSERT_NO_THROW(root_client->update_user(make_string_identifier(username), true, replacement, true,
                                             iggy::ffi::UserStatus::Active));
    RenameTrackedUser(username, replacement);
    iggy::ffi::UserInfoDetails fetched{};
    ASSERT_NO_THROW({ fetched = root_client->get_user(make_string_identifier(replacement)); });
    EXPECT_EQ(fetched.id, created.id);
    EXPECT_EQ(fetched.created_at, created.created_at);
    EXPECT_TRUE(fetched.has_permissions);
    EXPECT_TRUE(fetched.permissions.global.read_users);
    EXPECT_TRUE(fetched.permissions.global.read_streams);
    EXPECT_TRUE(fetched.permissions.global.send_messages);
    ASSERT_NO_THROW(valid_client->connect());
    ASSERT_NO_THROW(valid_client->login_user(replacement, "known-secret"));
    ASSERT_NO_THROW(old_client->connect());
    ASSERT_THROW(old_client->login_user(username, "known-secret"), std::exception);
}

TEST_F(LowLevelE2E_Client, UpdateUserToInactiveBlocksFreshLoginUntilReactivated) {
    RecordProperty("description", "Blocks fresh login while inactive and restores it when active.");
    iggy::ffi::Client *root_client = GetLoggedInClient();
    iggy::ffi::Client *user_client = GetLoggedOutClient();
    const std::string username     = GetRandomName(50);
    ASSERT_NO_THROW({ CreateUser(root_client, username, "known-secret", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW(root_client->update_user(make_string_identifier(username), true, username, true,
                                             iggy::ffi::UserStatus::Inactive));
    ASSERT_NO_THROW(user_client->connect());
    ASSERT_THROW(user_client->login_user(username, "known-secret"), std::exception);
    ASSERT_NO_THROW(root_client->update_user(make_string_identifier(username), true, username, true,
                                             iggy::ffi::UserStatus::Active));
    ASSERT_NO_THROW(user_client->login_user(username, "known-secret"));
}

TEST_F(LowLevelE2E_Client, UpdateUserRenameMakesOldUsernameReusable) {
    RecordProperty("description", "Releases the old username for a new user after a successful rename.");
    iggy::ffi::Client *client      = GetLoggedInClient();
    const std::string old_username = GetRandomName(50);
    const std::string new_username = GetRandomName(50);
    iggy::ffi::UserInfoDetails first{};
    ASSERT_NO_THROW({ first = CreateUser(client, old_username, "secret123", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW(client->update_user(make_string_identifier(old_username), true, new_username, true,
                                        iggy::ffi::UserStatus::Active));
    RenameTrackedUser(old_username, new_username);
    iggy::ffi::UserInfoDetails second{};
    ASSERT_NO_THROW({ second = CreateUser(client, old_username, "secret123", iggy::ffi::UserStatus::Active); });
    EXPECT_EQ(static_cast<std::string>(client->get_user(make_string_identifier(new_username)).username), new_username);
    EXPECT_EQ(static_cast<std::string>(client->get_user(make_string_identifier(old_username)).username), old_username);
}

TEST_F(LowLevelE2E_Client, UpdateUserRejectsRenameToRootUsername) {
    RecordProperty("description", "Rejects changing a non-root user's username to root's username.");
    iggy::ffi::Client *client  = GetLoggedInClient();
    const std::string username = GetRandomName(50);
    iggy::ffi::UserInfoDetails target{};
    ASSERT_NO_THROW({ target = CreateUser(client, username, "secret123", iggy::ffi::UserStatus::Active); });
    ASSERT_THROW(
        client->update_user(make_string_identifier(username), true, "iggy", true, iggy::ffi::UserStatus::Inactive),
        std::exception);
    iggy::ffi::UserInfoDetails root{};
    iggy::ffi::UserInfoDetails fetched{};
    ASSERT_NO_THROW({ root = client->get_user(make_string_identifier("iggy")); });
    ASSERT_NO_THROW({ fetched = client->get_user(make_string_identifier(username)); });
    EXPECT_EQ(root.id, 0u);
    EXPECT_EQ(static_cast<std::string>(root.username), "iggy");
    EXPECT_EQ(root.status, iggy::ffi::UserStatus::Active);
    EXPECT_EQ(fetched.id, target.id);
    EXPECT_EQ(fetched.status, iggy::ffi::UserStatus::Active);
    EXPECT_EQ(static_cast<std::string>(fetched.username), username);
}

TEST_F(LowLevelE2E_Client, ReadUsersPermissionDoesNotAllowUpdateUser) {
    RecordProperty("description", "Rejects updates from a user with read_users but not manage_users.");
    iggy::ffi::Client *root_client    = GetLoggedInClient();
    iggy::ffi::Client *caller_client  = GetLoggedOutClient();
    const std::string caller_username = GetRandomName(50);
    const std::string target_username = GetRandomName(50);
    iggy::ffi::Permissions permissions{};
    permissions.global.read_users = true;
    ASSERT_NO_THROW({
        CreateUser(root_client, caller_username, "secret123", iggy::ffi::UserStatus::Active, true,
                   std::move(permissions));
    });
    iggy::ffi::UserInfoDetails target{};
    ASSERT_NO_THROW({ target = CreateUser(root_client, target_username, "secret123", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW(caller_client->connect());
    ASSERT_NO_THROW(caller_client->login_user(caller_username, "secret123"));
    ASSERT_THROW(caller_client->update_user(make_string_identifier(target_username), true, GetRandomName(50), true,
                                            iggy::ffi::UserStatus::Inactive),
                 std::exception);
    ASSERT_THROW(caller_client->update_user(make_string_identifier(caller_username), true, GetRandomName(50), true,
                                            iggy::ffi::UserStatus::Inactive),
                 std::exception);
    iggy::ffi::UserInfoDetails fetched{};
    ASSERT_NO_THROW({ fetched = root_client->get_user(make_string_identifier(target_username)); });
    EXPECT_EQ(fetched.id, target.id);
    EXPECT_EQ(fetched.status, iggy::ffi::UserStatus::Active);
}

TEST_F(LowLevelE2E_Client, ManageUsersPermissionAllowsUpdateWithoutReadUsers) {
    RecordProperty("description", "Allows updates from a manager without read_users permission.");
    iggy::ffi::Client *root_client     = GetLoggedInClient();
    iggy::ffi::Client *manager_client  = GetLoggedOutClient();
    const std::string manager_username = GetRandomName(50);
    const std::string target_username  = GetRandomName(50);
    const std::string replacement      = GetRandomName(50);
    iggy::ffi::Permissions permissions{};
    permissions.global.manage_users = true;
    ASSERT_NO_THROW({
        CreateUser(root_client, manager_username, "secret123", iggy::ffi::UserStatus::Active, true,
                   std::move(permissions));
    });
    iggy::ffi::UserInfoDetails target{};
    ASSERT_NO_THROW({ target = CreateUser(root_client, target_username, "secret123", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW(manager_client->connect());
    ASSERT_NO_THROW(manager_client->login_user(manager_username, "secret123"));
    ASSERT_NO_THROW(manager_client->update_user(make_string_identifier(target_username), true, replacement, true,
                                                iggy::ffi::UserStatus::Inactive));
    RenameTrackedUser(target_username, replacement);
    iggy::ffi::UserInfoDetails fetched{};
    ASSERT_NO_THROW({ fetched = root_client->get_user(make_string_identifier(replacement)); });
    EXPECT_EQ(fetched.id, target.id);
    EXPECT_EQ(fetched.status, iggy::ffi::UserStatus::Inactive);
    EXPECT_EQ(static_cast<std::string>(fetched.username), replacement);
}

TEST_F(LowLevelE2E_Client, GetUserBeforeLoginThrows) {
    RecordProperty("description", "Rejects user lookup without an active authenticated session.");
    iggy::ffi::Client *client = GetLoggedOutClient();

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

TEST_F(LowLevelE2E_Client, GetUserByUsernameReturnsRootDetails) {
    RecordProperty("description", "Returns deterministic root details for a username lookup.");
    iggy::ffi::Client *client = GetLoggedInClient();

    iggy::ffi::UserInfoDetails user{};
    ASSERT_NO_THROW({ user = client->get_user(make_string_identifier("iggy")); });

    EXPECT_EQ(user.id, 0u);
    EXPECT_EQ(static_cast<std::string>(user.username), "iggy");
    EXPECT_EQ(user.status, iggy::ffi::UserStatus::Active);
}

TEST_F(LowLevelE2E_Client, GetUserByNumericIdMatchesUsernameLookup) {
    RecordProperty("description", "Returns equivalent root details for username and numeric identifiers.");
    iggy::ffi::Client *client = GetLoggedInClient();

    iggy::ffi::UserInfoDetails by_username{};
    iggy::ffi::UserInfoDetails by_id{};
    ASSERT_NO_THROW({ by_username = client->get_user(make_string_identifier("iggy")); });
    ASSERT_NO_THROW({ by_id = client->get_user(make_numeric_identifier(0)); });

    EXPECT_EQ(by_id.id, by_username.id);
    EXPECT_EQ(by_id.status, by_username.status);
    EXPECT_EQ(static_cast<std::string>(by_id.username), static_cast<std::string>(by_username.username));
    EXPECT_EQ(by_id.permissions.global.manage_servers, by_username.permissions.global.manage_servers);
    EXPECT_EQ(by_id.permissions.global.read_servers, by_username.permissions.global.read_servers);
    EXPECT_EQ(by_id.permissions.global.manage_users, by_username.permissions.global.manage_users);
    EXPECT_EQ(by_id.permissions.global.read_users, by_username.permissions.global.read_users);
    EXPECT_EQ(by_id.permissions.global.manage_streams, by_username.permissions.global.manage_streams);
    EXPECT_EQ(by_id.permissions.global.read_streams, by_username.permissions.global.read_streams);
    EXPECT_EQ(by_id.permissions.global.manage_topics, by_username.permissions.global.manage_topics);
    EXPECT_EQ(by_id.permissions.global.read_topics, by_username.permissions.global.read_topics);
    EXPECT_EQ(by_id.permissions.global.poll_messages, by_username.permissions.global.poll_messages);
    EXPECT_EQ(by_id.permissions.global.send_messages, by_username.permissions.global.send_messages);
    EXPECT_EQ(by_id.permissions.streams.size(), by_username.permissions.streams.size());
}

TEST_F(LowLevelE2E_Client, GetUserWithUnknownIdentifierThrows) {
    RecordProperty("description", "Rejects lookups for unknown username and numeric identifiers.");
    iggy::ffi::Client *client      = GetLoggedInClient();
    const std::string unknown_user = GetRandomName(50);

    ASSERT_THROW(client->get_user(make_string_identifier(unknown_user)), std::exception);
    ASSERT_THROW(client->get_user(make_numeric_identifier(std::numeric_limits<std::uint32_t>::max())), std::exception);
}

TEST_F(LowLevelE2E_Client, GetUsersBeforeLoginThrows) {
    RecordProperty("description", "Rejects listing users without an active authenticated session.");
    iggy::ffi::Client *client = GetLoggedOutClient();

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

TEST_F(LowLevelE2E_Client, GetUserAllowsSelfLookupWithoutReadUsersPermission) {
    RecordProperty("description", "Allows a user without read_users permission to read its own details.");
    iggy::ffi::Client *root_client = GetLoggedInClient();
    const std::string username     = GetRandomName(50);
    iggy::ffi::Permissions permissions{};
    permissions.global.read_users = false;
    iggy::ffi::UserInfoDetails created_user{};
    ASSERT_NO_THROW({
        created_user =
            CreateUser(root_client, username, "secret123", iggy::ffi::UserStatus::Active, true, std::move(permissions));
    });

    iggy::ffi::Client *user_client = GetLoggedOutClient();
    ASSERT_NO_THROW(user_client->connect());
    ASSERT_NO_THROW(user_client->login_user(username, "secret123"));

    iggy::ffi::UserInfoDetails by_username{};
    iggy::ffi::UserInfoDetails by_id{};
    ASSERT_NO_THROW({ by_username = user_client->get_user(make_string_identifier(username)); });
    ASSERT_NO_THROW({ by_id = user_client->get_user(make_numeric_identifier(created_user.id)); });

    EXPECT_EQ(by_username.id, created_user.id);
    EXPECT_EQ(by_id.id, created_user.id);
    EXPECT_EQ(static_cast<std::string>(by_username.username), username);
    EXPECT_EQ(static_cast<std::string>(by_id.username), username);
    EXPECT_EQ(by_username.status, iggy::ffi::UserStatus::Active);
    EXPECT_EQ(by_id.status, iggy::ffi::UserStatus::Active);
    EXPECT_FALSE(by_username.permissions.global.read_users);
    EXPECT_FALSE(by_id.permissions.global.read_users);
}

TEST_F(LowLevelE2E_Client, UserWithoutReadUsersPermissionCannotQueryOtherUsers) {
    RecordProperty("description", "Rejects other-user lookup and listing without read_users permission.");
    iggy::ffi::Client *root_client = GetLoggedInClient();
    const std::string username     = GetRandomName(50);
    iggy::ffi::Permissions permissions{};
    permissions.global.read_users = false;
    ASSERT_NO_THROW({
        CreateUser(root_client, username, "secret123", iggy::ffi::UserStatus::Active, true, std::move(permissions));
    });

    iggy::ffi::Client *user_client = GetLoggedOutClient();
    ASSERT_NO_THROW(user_client->connect());
    ASSERT_NO_THROW(user_client->login_user(username, "secret123"));

    ASSERT_THROW(user_client->get_user(make_numeric_identifier(0)), std::exception);
    ASSERT_THROW(user_client->get_users(), std::exception);
}

TEST_F(LowLevelE2E_Client, ReadUsersPermissionAllowsUserQueries) {
    RecordProperty("description", "Allows user lookup and listing with only read_users permission.");
    iggy::ffi::Client *root_client = GetLoggedInClient();
    const std::string username     = GetRandomName(50);
    iggy::ffi::Permissions permissions{};
    permissions.global.read_users = true;
    ASSERT_NO_THROW({
        CreateUser(root_client, username, "secret123", iggy::ffi::UserStatus::Active, true, std::move(permissions));
    });

    iggy::ffi::Client *user_client = GetLoggedOutClient();
    ASSERT_NO_THROW(user_client->connect());
    ASSERT_NO_THROW(user_client->login_user(username, "secret123"));

    iggy::ffi::UserInfoDetails root{};
    rust::Vec<iggy::ffi::UserInfo> users;
    ASSERT_NO_THROW({ root = user_client->get_user(make_numeric_identifier(0)); });
    ASSERT_NO_THROW({ users = user_client->get_users(); });

    EXPECT_EQ(root.id, 0u);
    EXPECT_EQ(static_cast<std::string>(root.username), "iggy");
    bool found_self = false;
    for (const auto &user : users) {
        if (static_cast<std::string>(user.username) == username) {
            found_self = true;
        }
    }
    EXPECT_TRUE(found_self);
}

TEST_F(LowLevelE2E_Client, ManageUsersPermissionImpliesReadUsers) {
    RecordProperty("description", "Allows user lookup and listing when manage_users is set without read_users.");
    iggy::ffi::Client *root_client = GetLoggedInClient();
    const std::string username     = GetRandomName(50);
    iggy::ffi::Permissions permissions{};
    permissions.global.manage_users = true;
    permissions.global.read_users   = false;
    ASSERT_NO_THROW({
        CreateUser(root_client, username, "secret123", iggy::ffi::UserStatus::Active, true, std::move(permissions));
    });

    iggy::ffi::Client *user_client = GetLoggedOutClient();
    ASSERT_NO_THROW(user_client->connect());
    ASSERT_NO_THROW(user_client->login_user(username, "secret123"));

    iggy::ffi::UserInfoDetails root{};
    rust::Vec<iggy::ffi::UserInfo> users;
    ASSERT_NO_THROW({ root = user_client->get_user(make_numeric_identifier(0)); });
    ASSERT_NO_THROW({ users = user_client->get_users(); });

    EXPECT_EQ(root.id, 0u);
    EXPECT_EQ(static_cast<std::string>(root.username), "iggy");
    bool found_self = false;
    for (const auto &user : users) {
        if (static_cast<std::string>(user.username) == username) {
            found_self = true;
        }
    }
    EXPECT_TRUE(found_self);
}

TEST_F(LowLevelE2E_Client, GetUsersContainsCreatedUser) {
    RecordProperty("description", "Lists a created user with the input username and status.");
    iggy::ffi::Client *client  = GetLoggedInClient();
    const std::string username = GetRandomName(50);
    iggy::ffi::UserInfoDetails created_user{};
    ASSERT_NO_THROW({ created_user = CreateUser(client, username, "secret123", iggy::ffi::UserStatus::Active); });

    rust::Vec<iggy::ffi::UserInfo> users;
    ASSERT_NO_THROW({ users = client->get_users(); });

    bool found_user = false;
    for (const auto &user : users) {
        if (user.id == created_user.id) {
            found_user = true;
            EXPECT_EQ(static_cast<std::string>(user.username), username);
            EXPECT_EQ(user.status, iggy::ffi::UserStatus::Active);
        }
    }
    EXPECT_TRUE(found_user);
}

TEST_F(LowLevelE2E_Client, GetUsersReturnsAllCreatedUsersOrderedById) {
    RecordProperty("description", "Returns every created user without pagination, ordered by ID over TCP.");
    iggy::ffi::Client *client = GetLoggedInClient();
    rust::Vec<iggy::ffi::UserInfo> users_before;
    ASSERT_NO_THROW({ users_before = client->get_users(); });

    const std::string first_username  = GetRandomName(50);
    const std::string second_username = GetRandomName(50);
    const std::string third_username  = GetRandomName(50);
    iggy::ffi::UserInfoDetails first_user{};
    iggy::ffi::UserInfoDetails second_user{};
    iggy::ffi::UserInfoDetails third_user{};
    ASSERT_NO_THROW({ first_user = CreateUser(client, first_username, "secret123", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW({ second_user = CreateUser(client, second_username, "secret123", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW({ third_user = CreateUser(client, third_username, "secret123", iggy::ffi::UserStatus::Active); });

    rust::Vec<iggy::ffi::UserInfo> users_after;
    ASSERT_NO_THROW({ users_after = client->get_users(); });
    ASSERT_EQ(users_after.size(), users_before.size() + 3);

    bool found_first  = false;
    bool found_second = false;
    bool found_third  = false;
    for (std::size_t index = 0; index < users_after.size(); ++index) {
        if (index > 0) {
            EXPECT_LT(users_after[index - 1].id, users_after[index].id);
        }
        const auto &user           = users_after[index];
        const std::string username = static_cast<std::string>(user.username);
        if (user.id == first_user.id && username == first_username) {
            found_first = true;
        }
        if (user.id == second_user.id && username == second_username) {
            found_second = true;
        }
        if (user.id == third_user.id && username == third_username) {
            found_third = true;
        }
    }
    EXPECT_TRUE(found_first);
    EXPECT_TRUE(found_second);
    EXPECT_TRUE(found_third);
}

TEST_F(LowLevelE2E_Client, GetUsersMatchesGetUserDetails) {
    RecordProperty("description", "Returns consistent ID, username, and status from user query APIs.");
    iggy::ffi::Client *client  = GetLoggedInClient();
    const std::string username = GetRandomName(50);
    iggy::ffi::UserInfoDetails created_user{};
    ASSERT_NO_THROW({ created_user = CreateUser(client, username, "secret123", iggy::ffi::UserStatus::Active); });

    iggy::ffi::UserInfoDetails user_details{};
    rust::Vec<iggy::ffi::UserInfo> users;
    ASSERT_NO_THROW({ user_details = client->get_user(make_string_identifier(username)); });
    ASSERT_NO_THROW({ users = client->get_users(); });

    EXPECT_EQ(user_details.id, created_user.id);
    EXPECT_EQ(static_cast<std::string>(user_details.username), username);
    EXPECT_EQ(user_details.status, iggy::ffi::UserStatus::Active);
    bool found_user = false;
    for (const auto &user : users) {
        if (user.id == user_details.id) {
            found_user = true;
            EXPECT_EQ(static_cast<std::string>(user.username), username);
            EXPECT_EQ(user.status, iggy::ffi::UserStatus::Active);
        }
    }
    EXPECT_TRUE(found_user);
}

TEST_F(LowLevelE2E_Client, DeletedUserDisappearsFromGetUserAndGetUsers) {
    RecordProperty("description", "Removes a deleted user from detail and list queries.");
    iggy::ffi::Client *client  = GetLoggedInClient();
    const std::string username = GetRandomName(50);
    iggy::ffi::UserInfoDetails created_user{};
    ASSERT_NO_THROW({ created_user = CreateUser(client, username, "secret123", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW(client->delete_user(make_string_identifier(username)));
    ForgetUser(username);

    ASSERT_THROW(client->get_user(make_string_identifier(username)), std::exception);
    rust::Vec<iggy::ffi::UserInfo> users;
    ASSERT_NO_THROW({ users = client->get_users(); });

    for (const auto &user : users) {
        EXPECT_FALSE(user.id == created_user.id && static_cast<std::string>(user.username) == username);
    }
}

TEST_F(LowLevelE2E_Client, DeleteUserRejectsUnauthenticatedClientWithoutDeletingTarget) {
    RecordProperty("description", "Rejects user deletion without an active authenticated session.");
    iggy::ffi::Client *root_client = GetLoggedInClient();
    iggy::ffi::Client *client      = GetLoggedOutClient();
    const std::string username     = GetRandomName(50);
    iggy::ffi::UserInfoDetails created_user{};
    ASSERT_NO_THROW({ created_user = CreateUser(root_client, username, "secret123", iggy::ffi::UserStatus::Active); });

    ASSERT_THROW(client->delete_user(make_string_identifier(username)), std::exception);
    ASSERT_NO_THROW(client->connect());
    ASSERT_THROW(client->delete_user(make_string_identifier(username)), std::exception);
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->logout_user());
    ASSERT_THROW(client->delete_user(make_string_identifier(username)), std::exception);
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->disconnect());
    ASSERT_THROW(client->delete_user(make_string_identifier(username)), std::exception);

    iggy::ffi::UserInfoDetails fetched_user{};
    ASSERT_NO_THROW({ fetched_user = root_client->get_user(make_string_identifier(username)); });
    EXPECT_EQ(fetched_user.id, created_user.id);
}

TEST_F(LowLevelE2E_Client, DeleteUserRejectsUnknownUsernameAndNumericId) {
    RecordProperty("description", "Rejects deletion of unknown string and numeric user identifiers.");
    iggy::ffi::Client *client          = GetLoggedInClient();
    const std::string unknown_user     = GetRandomName(50);
    constexpr std::uint32_t unknown_id = std::numeric_limits<std::uint32_t>::max();

    ASSERT_THROW(client->delete_user(make_string_identifier(unknown_user)), std::exception);
    ASSERT_THROW(client->delete_user(make_numeric_identifier(unknown_id)), std::exception);
}

TEST_F(LowLevelE2E_Client, DeleteUserRejectsRootWithoutChangingRoot) {
    RecordProperty("description", "Rejects root deletion by username and numeric ID without changing root.");
    iggy::ffi::Client *client = GetLoggedInClient();
    iggy::ffi::UserInfoDetails before_by_username{};
    iggy::ffi::UserInfoDetails before_by_id{};
    ASSERT_NO_THROW({ before_by_username = client->get_user(make_string_identifier("iggy")); });
    ASSERT_NO_THROW({ before_by_id = client->get_user(make_numeric_identifier(0)); });
    ASSERT_EQ(before_by_username.id, 0u);
    ASSERT_EQ(before_by_id.id, 0u);

    ASSERT_THROW(client->delete_user(make_string_identifier("iggy")), std::exception);
    ASSERT_THROW(client->delete_user(make_numeric_identifier(0)), std::exception);

    iggy::ffi::UserInfoDetails after_by_username{};
    iggy::ffi::UserInfoDetails after_by_id{};
    ASSERT_NO_THROW({ after_by_username = client->get_user(make_string_identifier("iggy")); });
    ASSERT_NO_THROW({ after_by_id = client->get_user(make_numeric_identifier(0)); });
    for (const auto *root : {&after_by_username, &after_by_id}) {
        EXPECT_EQ(root->id, 0u);
        EXPECT_EQ(static_cast<std::string>(root->username), "iggy");
        EXPECT_EQ(root->status, iggy::ffi::UserStatus::Active);
    }
}

TEST_F(LowLevelE2E_Client, DeleteUserByNumericIdRemovesTarget) {
    RecordProperty("description", "Deletes a user selected by its numeric ID.");
    iggy::ffi::Client *client  = GetLoggedInClient();
    const std::string username = GetRandomName(50);
    iggy::ffi::UserInfoDetails created_user{};
    ASSERT_NO_THROW({ created_user = CreateUser(client, username, "secret123", iggy::ffi::UserStatus::Active); });

    ASSERT_NO_THROW(client->delete_user(make_numeric_identifier(created_user.id)));
    ForgetUser(username);

    ASSERT_THROW(client->get_user(make_string_identifier(username)), std::exception);
    rust::Vec<iggy::ffi::UserInfo> users;
    ASSERT_NO_THROW({ users = client->get_users(); });
    for (const auto &user : users) {
        EXPECT_FALSE(user.id == created_user.id && static_cast<std::string>(user.username) == username);
    }
}

TEST_F(LowLevelE2E_Client, DeleteUserRejectsRepeatedDeletion) {
    RecordProperty("description", "Rejects deleting the same user twice.");
    iggy::ffi::Client *client  = GetLoggedInClient();
    const std::string username = GetRandomName(50);
    iggy::ffi::UserInfoDetails created_user{};
    ASSERT_NO_THROW({ created_user = CreateUser(client, username, "secret123", iggy::ffi::UserStatus::Active); });

    ASSERT_NO_THROW(client->delete_user(make_string_identifier(username)));
    ForgetUser(username);
    ASSERT_THROW(client->delete_user(make_string_identifier(username)), std::exception);

    rust::Vec<iggy::ffi::UserInfo> users;
    ASSERT_NO_THROW({ users = client->get_users(); });
    for (const auto &user : users) {
        EXPECT_FALSE(user.id == created_user.id && static_cast<std::string>(user.username) == username);
    }
}

TEST_F(LowLevelE2E_Client, ReadUsersPermissionDoesNotAllowDeleteUser) {
    RecordProperty("description", "Rejects other-user and self deletion with read_users but not manage_users.");
    iggy::ffi::Client *root_client   = GetLoggedInClient();
    iggy::ffi::Client *caller_client = GetLoggedOutClient();
    const std::string caller_name    = GetRandomName(50);
    const std::string target_name    = GetRandomName(50);
    iggy::ffi::Permissions permissions{};
    permissions.global.read_users   = true;
    permissions.global.manage_users = false;
    iggy::ffi::UserInfoDetails caller{};
    iggy::ffi::UserInfoDetails target{};
    ASSERT_NO_THROW({
        caller = CreateUser(root_client, caller_name, "caller-secret", iggy::ffi::UserStatus::Active, true,
                            std::move(permissions));
    });
    ASSERT_NO_THROW({ target = CreateUser(root_client, target_name, "target-secret", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW(caller_client->connect());
    ASSERT_NO_THROW(caller_client->login_user(caller_name, "caller-secret"));

    ASSERT_THROW(caller_client->delete_user(make_string_identifier(target_name)), std::exception);
    ASSERT_THROW(caller_client->delete_user(make_string_identifier(caller_name)), std::exception);

    iggy::ffi::UserInfoDetails fetched_caller{};
    iggy::ffi::UserInfoDetails fetched_target{};
    ASSERT_NO_THROW({ fetched_caller = root_client->get_user(make_string_identifier(caller_name)); });
    ASSERT_NO_THROW({ fetched_target = root_client->get_user(make_string_identifier(target_name)); });
    EXPECT_EQ(fetched_caller.id, caller.id);
    EXPECT_EQ(fetched_target.id, target.id);
}

TEST_F(LowLevelE2E_Client, ManageUsersPermissionAllowsDeleteUser) {
    RecordProperty("description", "Allows deletion with manage_users even when read_users is false.");
    iggy::ffi::Client *root_client    = GetLoggedInClient();
    iggy::ffi::Client *manager_client = GetLoggedOutClient();
    const std::string manager_name    = GetRandomName(50);
    const std::string target_name     = GetRandomName(50);
    iggy::ffi::Permissions permissions{};
    permissions.global.manage_users = true;
    permissions.global.read_users   = false;
    iggy::ffi::UserInfoDetails target{};
    ASSERT_NO_THROW({
        CreateUser(root_client, manager_name, "manager-secret", iggy::ffi::UserStatus::Active, true,
                   std::move(permissions));
    });
    ASSERT_NO_THROW({ target = CreateUser(root_client, target_name, "target-secret", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW(manager_client->connect());
    ASSERT_NO_THROW(manager_client->login_user(manager_name, "manager-secret"));

    ASSERT_NO_THROW(manager_client->delete_user(make_string_identifier(target_name)));
    ForgetUser(target_name);

    ASSERT_THROW(root_client->get_user(make_string_identifier(target_name)), std::exception);
    rust::Vec<iggy::ffi::UserInfo> users;
    ASSERT_NO_THROW({ users = root_client->get_users(); });
    for (const auto &user : users) {
        EXPECT_FALSE(user.id == target.id && static_cast<std::string>(user.username) == target_name);
    }
}

TEST_F(LowLevelE2E_Client, DeleteUserRemovesOnlyTheTarget) {
    RecordProperty("description", "Deletes only the selected user and leaves another user usable.");
    iggy::ffi::Client *root_client     = GetLoggedInClient();
    iggy::ffi::Client *survivor_client = GetLoggedOutClient();
    const std::string target_name      = GetRandomName(50);
    const std::string survivor_name    = GetRandomName(50);
    iggy::ffi::UserInfoDetails target{};
    iggy::ffi::UserInfoDetails survivor{};
    ASSERT_NO_THROW({ target = CreateUser(root_client, target_name, "target-secret", iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW(
        { survivor = CreateUser(root_client, survivor_name, "survivor-secret", iggy::ffi::UserStatus::Active); });

    ASSERT_NO_THROW(root_client->delete_user(make_numeric_identifier(target.id)));
    ForgetUser(target_name);

    ASSERT_THROW(root_client->get_user(make_string_identifier(target_name)), std::exception);
    rust::Vec<iggy::ffi::UserInfo> users;
    ASSERT_NO_THROW({ users = root_client->get_users(); });
    for (const auto &user : users) {
        EXPECT_FALSE(user.id == target.id && static_cast<std::string>(user.username) == target_name);
    }

    iggy::ffi::UserInfoDetails fetched_survivor{};
    ASSERT_NO_THROW({ fetched_survivor = root_client->get_user(make_string_identifier(survivor_name)); });
    EXPECT_EQ(fetched_survivor.id, survivor.id);
    EXPECT_EQ(static_cast<std::string>(fetched_survivor.username), survivor_name);
    EXPECT_EQ(fetched_survivor.status, survivor.status);
    ASSERT_NO_THROW(survivor_client->connect());
    ASSERT_NO_THROW(survivor_client->login_user(survivor_name, "survivor-secret"));
}

TEST_F(LowLevelE2E_Client, DeleteInactiveUserSucceeds) {
    RecordProperty("description", "Deletes an inactive user by username.");
    iggy::ffi::Client *client  = GetLoggedInClient();
    const std::string username = GetRandomName(50);
    iggy::ffi::UserInfoDetails created_user{};
    ASSERT_NO_THROW({ created_user = CreateUser(client, username, "secret123", iggy::ffi::UserStatus::Inactive); });

    ASSERT_NO_THROW(client->delete_user(make_string_identifier(username)));
    ForgetUser(username);

    ASSERT_THROW(client->get_user(make_string_identifier(username)), std::exception);
    rust::Vec<iggy::ffi::UserInfo> users;
    ASSERT_NO_THROW({ users = client->get_users(); });
    for (const auto &user : users) {
        EXPECT_FALSE(user.id == created_user.id && static_cast<std::string>(user.username) == username);
    }
}

TEST_F(LowLevelE2E_Client, DeleteUserRevokesExistingSessions) {
    RecordProperty("description", "Revokes an existing session and prevents fresh login after user deletion.");
    iggy::ffi::Client *root_client   = GetLoggedInClient();
    iggy::ffi::Client *target_client = GetLoggedOutClient();
    iggy::ffi::Client *fresh_client  = GetLoggedOutClient();
    const std::string username       = GetRandomName(50);
    const std::string password       = "target-secret";
    iggy::ffi::Permissions permissions{};
    permissions.global.read_servers = true;
    ASSERT_NO_THROW(
        { CreateUser(root_client, username, password, iggy::ffi::UserStatus::Active, true, std::move(permissions)); });
    ASSERT_NO_THROW(target_client->connect());
    ASSERT_NO_THROW(target_client->login_user(username, password));
    iggy::ffi::Stats stats{};
    ASSERT_NO_THROW({ stats = target_client->get_stats(); });

    ASSERT_NO_THROW(root_client->delete_user(make_string_identifier(username)));
    ForgetUser(username);

    ASSERT_THROW(target_client->get_stats(), std::exception);
    ASSERT_NO_THROW(fresh_client->connect());
    ASSERT_THROW(fresh_client->login_user(username, password), std::exception);
}

TEST_F(LowLevelE2E_Client, DeletedUsernameCanBeRecreatedWithoutOldCredentialsOrPermissions) {
    RecordProperty("description", "Recreates a deleted username without retaining old credentials or permissions.");
    iggy::ffi::Client *root_client         = GetLoggedInClient();
    iggy::ffi::Client *new_password_client = GetLoggedOutClient();
    iggy::ffi::Client *old_password_client = GetLoggedOutClient();
    const std::string username             = GetRandomName(50);
    const std::string old_password         = "old-secret";
    const std::string new_password         = "new-secret";
    iggy::ffi::Permissions permissions{};
    permissions.global.read_servers = true;
    ASSERT_NO_THROW({
        CreateUser(root_client, username, old_password, iggy::ffi::UserStatus::Active, true, std::move(permissions));
    });

    ASSERT_NO_THROW(root_client->delete_user(make_string_identifier(username)));
    ForgetUser(username);

    iggy::ffi::UserInfoDetails replacement{};
    ASSERT_NO_THROW({ replacement = CreateUser(root_client, username, new_password, iggy::ffi::UserStatus::Active); });
    EXPECT_EQ(static_cast<std::string>(replacement.username), username);
    EXPECT_FALSE(replacement.has_permissions);
    EXPECT_FALSE(replacement.permissions.global.read_servers);
    EXPECT_TRUE(replacement.permissions.streams.empty());

    iggy::ffi::UserInfoDetails fetched_replacement{};
    ASSERT_NO_THROW({ fetched_replacement = root_client->get_user(make_string_identifier(username)); });
    EXPECT_EQ(fetched_replacement.id, replacement.id);

    ASSERT_NO_THROW(new_password_client->connect());
    ASSERT_NO_THROW(new_password_client->login_user(username, new_password));
    ASSERT_NO_THROW(old_password_client->connect());
    ASSERT_THROW(old_password_client->login_user(username, old_password), std::exception);
}

TEST_F(LowLevelE2E_Client, ChangePasswordBeforeLoginThrows) {
    RecordProperty("description",
                   "Rejects change_password before connect, after connect but before login, and after disconnect.");
    iggy::ffi::Client *client      = GetLoggedOutClient();
    const auto user_id             = make_string_identifier("iggy");
    const std::string old_password = "iggy";
    const std::string new_password = "iggy-updated-secret";

    ASSERT_THROW(client->change_password(user_id, old_password, new_password), std::exception);
    ASSERT_NO_THROW(client->connect());
    ASSERT_THROW(client->change_password(user_id, old_password, new_password), std::exception);
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->disconnect());
    ASSERT_THROW(client->change_password(user_id, old_password, new_password), std::exception);
}

TEST_F(LowLevelE2E_Client, ChangePasswordWithInvalidCurrentPasswordThrows) {
    RecordProperty("description", "Rejects change_password when the provided current password is incorrect.");
    iggy::ffi::Client *client        = GetLoggedInClient();
    const auto user_id               = make_string_identifier("iggy");
    const std::string wrong_password = "not-the-current-password";
    const std::string new_password   = "iggy-updated-secret";

    ASSERT_THROW(client->change_password(user_id, wrong_password, new_password), std::exception);
    ASSERT_NO_THROW(client->logout_user());
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
}

TEST_F(LowLevelE2E_Client, ChangePasswordWithInvalidNewPasswordThrows) {
    RecordProperty("description",
                   "Rejects change_password when the replacement password violates client-side length bounds.");
    iggy::ffi::Client *client      = GetLoggedInClient();
    const auto user_id             = make_string_identifier("iggy");
    const std::string old_password = "iggy";
    const std::string too_short    = "";
    const std::string too_long(256, 'a');

    ASSERT_THROW(client->change_password(user_id, old_password, too_short), std::exception);
    ASSERT_THROW(client->change_password(user_id, old_password, too_long), std::exception);
    ASSERT_NO_THROW(client->logout_user());
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
}

TEST_F(LowLevelE2E_Client, ChangePasswordForWrongUserThrows) {
    RecordProperty("description", "Rejects change_password when targeting a user that does not exist.");
    iggy::ffi::Client *client            = GetLoggedInClient();
    const auto wrong_user_id             = make_string_identifier(GetRandomName());
    const std::string current_password   = "iggy";
    const std::string replacement_secret = "iggy-updated-secret";

    ASSERT_THROW(client->change_password(wrong_user_id, current_password, replacement_secret), std::exception);
    ASSERT_NO_THROW(client->logout_user());
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
}

TEST_F(LowLevelE2E_Client, UserWithoutManageUsersCannotChangeAnotherUsersPassword) {
    RecordProperty("description",
                   "Rejects another user's password change without manage_users and preserves the old credentials.");
    iggy::ffi::Client *root_client    = GetLoggedInClient();
    iggy::ffi::Client *actor_client   = GetLoggedOutClient();
    iggy::ffi::Client *target_client  = GetLoggedOutClient();
    const std::string actor_name      = GetRandomName(50);
    const std::string target_name     = GetRandomName(50);
    const std::string target_password = "target-secret";
    const std::string new_password    = "replacement-secret";
    ASSERT_NO_THROW({ CreateUser(root_client, actor_name, "actor-secret", iggy::ffi::UserStatus::Active); });
    iggy::ffi::UserInfoDetails target{};
    ASSERT_NO_THROW({ target = CreateUser(root_client, target_name, target_password, iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW(actor_client->connect());
    ASSERT_NO_THROW(actor_client->login_user(actor_name, "actor-secret"));

    ASSERT_THROW(actor_client->change_password(make_numeric_identifier(target.id), target_password, new_password),
                 std::exception);

    ASSERT_NO_THROW(target_client->connect());
    ASSERT_NO_THROW(target_client->login_user(target_name, target_password));
}

TEST_F(LowLevelE2E_Client, ManageUsersPermissionAllowsChangingAnotherUsersPassword) {
    RecordProperty("description", "Allows a user with manage_users to change another user's password.");
    iggy::ffi::Client *root_client         = GetLoggedInClient();
    iggy::ffi::Client *manager_client      = GetLoggedOutClient();
    iggy::ffi::Client *old_password_client = GetLoggedOutClient();
    iggy::ffi::Client *new_password_client = GetLoggedOutClient();
    const std::string manager_name         = GetRandomName(50);
    const std::string target_name          = GetRandomName(50);
    const std::string target_password      = "target-secret";
    const std::string new_password         = "replacement-secret";
    iggy::ffi::Permissions permissions{};
    permissions.global.manage_users = true;
    ASSERT_NO_THROW({
        CreateUser(root_client, manager_name, "manager-secret", iggy::ffi::UserStatus::Active, true,
                   std::move(permissions));
    });
    iggy::ffi::UserInfoDetails target{};
    ASSERT_NO_THROW({ target = CreateUser(root_client, target_name, target_password, iggy::ffi::UserStatus::Active); });
    ASSERT_NO_THROW(manager_client->connect());
    ASSERT_NO_THROW(manager_client->login_user(manager_name, "manager-secret"));

    ASSERT_NO_THROW(manager_client->change_password(make_numeric_identifier(target.id), target_password, new_password));

    ASSERT_NO_THROW(old_password_client->connect());
    ASSERT_THROW(old_password_client->login_user(target_name, target_password), std::exception);
    ASSERT_NO_THROW(new_password_client->connect());
    ASSERT_NO_THROW(new_password_client->login_user(target_name, new_password));
}

TEST_F(LowLevelE2E_Client, ChangePasswordUpdatesCredentialsAndCanBeRestored) {
    RecordProperty("description",
                   "Changes the password for the current user, updates login behavior, and restores the original "
                   "password before the test exits.");
    iggy::ffi::Client *client        = GetLoggedInClient();
    iggy::ffi::Client *second_client = GetLoggedOutClient();
    iggy::ffi::Client *third_client  = GetLoggedOutClient();
    const auto user_id               = make_string_identifier("iggy");
    const std::string old_password   = "iggy";
    const std::string new_password   = "iggy-updated-secret";
    bool password_changed            = false;

    ASSERT_NO_THROW(client->change_password(user_id, old_password, new_password));
    password_changed = true;

    EXPECT_THROW(second_client->login_user("iggy", old_password), std::exception);
    EXPECT_NO_THROW(second_client->login_user("iggy", new_password));

    if (password_changed) {
        EXPECT_NO_THROW(client->change_password(user_id, new_password, old_password));
        password_changed = false;
    }

    EXPECT_NO_THROW(third_client->login_user("iggy", old_password));
}

TEST_F(LowLevelE2E_Client, DeleteWhileUnauthenticatedAfterFailedLogin) {
    RecordProperty("description", "Allows client cleanup after a failed login leaves the connection unauthenticated.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);

    ASSERT_NO_THROW(client->connect());
    ASSERT_THROW(client->login_user("biggy", "biggy"), std::exception);
    iggy::ffi::delete_client(client);
    client = nullptr;
}

TEST_F(LowLevelE2E_Client, ConnectLoginThenDisconnect) {
    RecordProperty("description",
                   "Connects, logs in, disconnects successfully, and rejects authenticated operations afterward.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

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

TEST_F(LowLevelE2E_Client, DisconnectWithoutConnect) {
    RecordProperty("description", "Allows disconnect to be called on a client that was never explicitly connected.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

    ASSERT_NO_THROW(client->disconnect());
}

TEST_F(LowLevelE2E_Client, DisconnectWithoutLogin) {
    RecordProperty("description", "Allows disconnect after connect even when no user has authenticated.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

    ASSERT_NO_THROW(client->connect());
    ASSERT_NO_THROW(client->disconnect());
    ASSERT_THROW(client->get_stats(), std::exception);
}

TEST_F(LowLevelE2E_Client, DisconnectThenReconnectWithoutRelogin) {
    RecordProperty("description",
                   "Requires logging in again after a disconnect and reconnect before authenticated operations work.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

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

TEST_F(LowLevelE2E_Client, DisconnectAfterFailedLogin) {
    RecordProperty("description", "Allows disconnect after a failed login attempt leaves the client unauthenticated.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

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

TEST_F(LowLevelE2E_Client, ConnectLoginThenShutdown) {
    RecordProperty("description",
                   "Connects, logs in, shuts down successfully, and rejects further operations afterward.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

    ASSERT_NO_THROW(client->connect());
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->ping());
    ASSERT_NO_THROW(client->shutdown());
    ASSERT_THROW(client->get_me(), std::exception);
    ASSERT_THROW(client->get_stats(), std::exception);
}

TEST_F(LowLevelE2E_Client, ShutdownWithoutConnect) {
    RecordProperty("description", "Allows shutdown to be called on a client that was never explicitly connected.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

    ASSERT_NO_THROW(client->shutdown());
}

TEST_F(LowLevelE2E_Client, ShutdownWithoutLogin) {
    RecordProperty("description", "Allows shutdown after connect even when no user has authenticated.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

    ASSERT_NO_THROW(client->connect());
    ASSERT_NO_THROW(client->shutdown());
    ASSERT_THROW(client->get_stats(), std::exception);
}

TEST_F(LowLevelE2E_Client, ShutdownAfterFailedLogin) {
    RecordProperty("description", "Allows shutdown after a failed login attempt leaves the client unauthenticated.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

    ASSERT_NO_THROW(client->connect());
    ASSERT_THROW(client->login_user("biggy", "biggy"), std::exception);
    ASSERT_NO_THROW(client->shutdown());
    ASSERT_THROW(client->get_me(), std::exception);
}

TEST_F(LowLevelE2E_Client, RepeatedShutdownCallsHaveStableBehavior) {
    RecordProperty("description", "Keeps repeated shutdown calls stable across duplicate invocations.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

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

TEST_F(LowLevelE2E_Client, ShutdownThenConnectThrows) {
    RecordProperty("description", "Rejects reconnecting a client after shutdown transitions it to a terminal state.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

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

TEST_F(LowLevelE2E_Client, ShutdownThenLoginThrows) {
    RecordProperty("description",
                   "Rejects logging in again after shutdown, even when login would normally auto-connect.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

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

TEST_F(LowLevelE2E_Client, GetClientsReflectsSessionRemovalAfterShutdown) {
    RecordProperty("description",
                   "Removes a shut down authenticated session from subsequent get_clients and get_client results.");
    iggy::ffi::Client *first_client  = GetLoggedInClient();
    iggy::ffi::Client *second_client = GetLoggedInClient();

    iggy::ffi::ClientInfoDetails first_me{};
    ASSERT_NO_THROW({ first_me = first_client->get_me(); });

    ASSERT_NO_THROW(first_client->shutdown());
    constexpr auto removal_timeout       = std::chrono::seconds(5);
    constexpr auto removal_poll_interval = std::chrono::milliseconds(10);
    const auto deadline                  = std::chrono::steady_clock::now() + removal_timeout;
    bool removed                         = false;
    do {
        const auto clients = second_client->get_clients();
        removed            = std::none_of(clients.begin(), clients.end(),
                                          [&first_me](const auto &client) { return client.client_id == first_me.client_id; });
        if (removed) {
            break;
        }
        std::this_thread::sleep_for(removal_poll_interval);
    } while (std::chrono::steady_clock::now() < deadline);
    ASSERT_TRUE(removed);
    ASSERT_THROW(second_client->get_client(first_me.client_id), std::exception);
}

TEST_F(LowLevelE2E_Client, GetClientsReflectsSessionRemovalAfterDisconnect) {
    RecordProperty("description",
                   "Removes a disconnected authenticated session from subsequent get_clients and get_client results.");
    iggy::ffi::Client *first_client  = GetLoggedInClient();
    iggy::ffi::Client *second_client = GetLoggedInClient();

    iggy::ffi::ClientInfoDetails first_me{};
    ASSERT_NO_THROW({ first_me = first_client->get_me(); });

    ASSERT_NO_THROW(first_client->disconnect());
    constexpr auto removal_timeout       = std::chrono::seconds(5);
    constexpr auto removal_poll_interval = std::chrono::milliseconds(10);
    const auto deadline                  = std::chrono::steady_clock::now() + removal_timeout;
    bool removed                         = false;
    do {
        const auto clients = second_client->get_clients();
        removed            = std::none_of(clients.begin(), clients.end(),
                                          [&first_me](const auto &client) { return client.client_id == first_me.client_id; });
        if (removed) {
            break;
        }
        std::this_thread::sleep_for(removal_poll_interval);
    } while (std::chrono::steady_clock::now() < deadline);
    ASSERT_TRUE(removed);
    ASSERT_THROW(second_client->get_client(first_me.client_id), std::exception);
}

TEST_F(LowLevelE2E_Client, GetClientsReflectsLoggedOutSessionAsUnauthenticated) {
    RecordProperty("description", "Drops a logged out session from get_clients and reports it missing in get_client.");
    iggy::ffi::Client *first_client  = GetLoggedInClient();
    iggy::ffi::Client *second_client = GetLoggedInClient();

    iggy::ffi::ClientInfoDetails first_me{};
    ASSERT_NO_THROW({ first_me = first_client->get_me(); });

    // The VSR server drops the client-table entry on logout (an unauthenticated
    // session is not tracked), unlike the legacy server which kept it visible
    // without a user id.
    ASSERT_NO_THROW(first_client->logout_user());
    constexpr auto removal_timeout       = std::chrono::seconds(5);
    constexpr auto removal_poll_interval = std::chrono::milliseconds(10);
    const auto deadline                  = std::chrono::steady_clock::now() + removal_timeout;
    bool removed                         = false;
    do {
        const auto clients = second_client->get_clients();
        removed            = std::none_of(clients.begin(), clients.end(),
                                          [&first_me](const auto &client) { return client.client_id == first_me.client_id; });
        if (removed) {
            break;
        }
        std::this_thread::sleep_for(removal_poll_interval);
    } while (std::chrono::steady_clock::now() < deadline);
    ASSERT_TRUE(removed);
    ASSERT_THROW(second_client->get_client(first_me.client_id), std::exception);
}

TEST_F(LowLevelE2E_Client, LoginWithoutConnect) {
    RecordProperty("description", "Supports login without an explicit prior connect call.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
}

TEST_F(LowLevelE2E_Client, ConnectWithoutLoginThenDelete) {
    RecordProperty("description", "Allows connecting without logging in and then deleting the client.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);

    ASSERT_NO_THROW(client->connect());
    iggy::ffi::delete_client(client);
    client = nullptr;
}

TEST_F(LowLevelE2E_Client, DeleteWithoutDisconnect) {
    RecordProperty("description", "Allows deleting a connected and authenticated client without disconnecting first.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);

    ASSERT_NO_THROW(client->connect());
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    iggy::ffi::delete_client(client);
    client = nullptr;
}

TEST_F(LowLevelE2E_Client, RepeatedClientMethodCallsHaveStableBehavior) {
    RecordProperty("description",
                   "Keeps repeated connect, login, and delete calls stable across duplicate invocations.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);

    ASSERT_NO_THROW(client->connect());
    ASSERT_NO_THROW(client->connect());
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    iggy::ffi::delete_client(client);
    client = nullptr;

    iggy::ffi::delete_client(client);
}

TEST_F(LowLevelE2E_Client, RepeatedDisconnectCallsHaveStableBehavior) {
    RecordProperty("description", "Keeps repeated disconnect calls stable across duplicate invocations.");
    iggy::ffi::Client *client = nullptr;
    ASSERT_NO_THROW({ client = iggy::ffi::new_connection({}); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

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

TEST_F(LowLevelE2E_Client, DeleteNullConnectionIsNoop) {
    RecordProperty("description", "Treats deleting a null client pointer as a no-op.");
    iggy::ffi::Client *client = nullptr;
    iggy::ffi::delete_client(client);
}

TEST_F(LowLevelE2E_Client, GetStatsBeforeLoginThrows) {
    RecordProperty("description",
                   "Rejects get_stats before connect, after connect but before login, and after disconnect.");
    iggy::ffi::Client *client = GetLoggedOutClient();

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

// The VSR server has no unsaved-buffer primitive (writes are journaled at
// commit); FLUSH_UNSAVED_BUFFER denies typed with FeatureUnavailable even for
// resolvable targets.
TEST_F(LowLevelE2E_Client, FlushUnsavedBufferThrowsForExistingPartition) {
    RecordProperty("description",
                   "Rejects flush_unsaved_buffer with the feature-unavailable error for an existing partition.");
    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));
    auto stream = client->get_stream(make_string_identifier(stream_name));
    TrackStream(stream.id);
    ASSERT_NO_THROW(client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0,
                                         "server_default", {}));

    rust::Vec<iggy::ffi::IggyMessageToSend> messages;
    messages.push_back(iggy::ffi::make_message(to_payload("flush-me"), rust::Vec<iggy::ffi::HeaderEntry>()));

    ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0),
                                          "partition_id", partition_id_bytes(0), std::move(messages)));
    ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, true),
                 std::exception);
}

TEST_F(LowLevelE2E_Client, FlushUnsavedBufferThrowsForExistingEmptyPartition) {
    RecordProperty(
        "description",
        "Rejects flush_unsaved_buffer with the feature-unavailable error for a partition with no unsaved messages.");
    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));
    auto stream = client->get_stream(make_string_identifier(stream_name));
    TrackStream(stream.id);
    ASSERT_NO_THROW(client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0,
                                         "server_default", {}));

    ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, true),
                 std::exception);
}

TEST_F(LowLevelE2E_Client, FlushUnsavedBufferBeforeLoginThrows) {
    RecordProperty("description",
                   "Throws when flush_unsaved_buffer is called before connect, after connect but before login, and "
                   "after disconnect.");
    iggy::ffi::Client *client = GetLoggedOutClient();

    ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(1), make_numeric_identifier(1), 0, true),
                 std::exception);
    ASSERT_NO_THROW(client->connect());
    ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(1), make_numeric_identifier(1), 0, true),
                 std::exception);
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->disconnect());
    ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(1), make_numeric_identifier(1), 0, true),
                 std::exception);
}

TEST_F(LowLevelE2E_Client, FlushUnsavedBufferOnNonExistentStreamThrows) {
    RecordProperty("description", "Throws when flush_unsaved_buffer is called for a stream that does not exist.");
    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", "never_expire", 0,
                                         "server_default", {}));

    ASSERT_THROW(
        client->flush_unsaved_buffer(make_string_identifier(GetRandomName()), make_numeric_identifier(0), 0, true),
        std::exception);
}

TEST_F(LowLevelE2E_Client, FlushUnsavedBufferOnNonExistentTopicThrows) {
    RecordProperty("description", "Throws when flush_unsaved_buffer is called for a topic that does not exist.");
    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", "never_expire", 0,
                                         "server_default", {}));

    ASSERT_THROW(client->flush_unsaved_buffer(make_string_identifier(stream_name),
                                              make_string_identifier(GetRandomName()), 0, true),
                 std::exception);
}

TEST_F(LowLevelE2E_Client, FlushUnsavedBufferAfterStreamDeletedThrows) {
    RecordProperty("description", "Throws when flush_unsaved_buffer 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));
    auto stream = client->get_stream(make_string_identifier(stream_name));
    TrackStream(stream.id);
    ASSERT_NO_THROW(client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0,
                                         "server_default", {}));

    const std::uint32_t saved_stream_id = stream.id;
    ASSERT_NO_THROW(client->delete_stream(make_numeric_identifier(saved_stream_id)));
    ForgetTrackedStream(saved_stream_id);

    ASSERT_THROW(
        client->flush_unsaved_buffer(make_numeric_identifier(saved_stream_id), make_numeric_identifier(0), 0, true),
        std::exception);
}

TEST_F(LowLevelE2E_Client, FlushUnsavedBufferAfterTopicDeletedThrows) {
    RecordProperty("description", "Throws when flush_unsaved_buffer is called after the topic 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));
    auto stream = client->get_stream(make_string_identifier(stream_name));
    TrackStream(stream.id);
    ASSERT_NO_THROW(client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0,
                                         "server_default", {}));
    ASSERT_NO_THROW(client->delete_topic(make_numeric_identifier(stream.id), make_string_identifier(topic_name)));

    ASSERT_THROW(
        client->flush_unsaved_buffer(make_numeric_identifier(stream.id), make_string_identifier(topic_name), 0, true),
        std::exception);
}

TEST_F(LowLevelE2E_Client, FlushUnsavedBufferTwiceThrows) {
    RecordProperty("description",
                   "Rejects flush_unsaved_buffer with the feature-unavailable error consistently across repeat 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));
    auto stream = client->get_stream(make_string_identifier(stream_name));
    TrackStream(stream.id);
    ASSERT_NO_THROW(client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0,
                                         "server_default", {}));

    rust::Vec<iggy::ffi::IggyMessageToSend> messages;
    messages.push_back(iggy::ffi::make_message(to_payload("flush-twice"), rust::Vec<iggy::ffi::HeaderEntry>()));

    ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0),
                                          "partition_id", partition_id_bytes(0), std::move(messages)));
    ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, true),
                 std::exception);
    ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, true),
                 std::exception);
}

TEST_F(LowLevelE2E_Client, FlushUnsavedBufferWithInvalidPartitionIdsThrows) {
    RecordProperty("description", "Throws when flush_unsaved_buffer is called for non-existent partition ids.");
    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));
    auto stream = client->get_stream(make_string_identifier(stream_name));
    TrackStream(stream.id);
    ASSERT_NO_THROW(client->create_topic(make_numeric_identifier(stream.id), topic_name, 1, "none", "never_expire", 0,
                                         "server_default", {}));

    const std::uint32_t invalid_partition_ids[] = {1u, 9999u, static_cast<std::uint32_t>(-1)};
    for (const std::uint32_t invalid_partition_id : invalid_partition_ids) {
        SCOPED_TRACE(invalid_partition_id);
        ASSERT_THROW(client->flush_unsaved_buffer(make_numeric_identifier(stream.id), make_numeric_identifier(0),
                                                  invalid_partition_id, true),
                     std::exception);
    }
}

TEST_F(LowLevelE2E_Client, DeleteSegmentsBeforeLoginThrows) {
    RecordProperty("description",
                   "Rejects delete_segments before connect, after connect but before login, and after disconnect.");
    const std::string stream_name   = GetRandomName();
    const std::string topic_name    = GetRandomName();
    iggy::ffi::Client *setup_client = GetLoggedInClient();

    ASSERT_NO_THROW(setup_client->create_stream(stream_name));
    TrackStream(stream_name);
    ASSERT_NO_THROW(setup_client->create_topic(make_string_identifier(stream_name), topic_name, 1, "none",
                                               "never_expire", 0, "server_default", {}));

    iggy::ffi::Client *unauthenticated_client = GetLoggedOutClient();
    ASSERT_THROW(unauthenticated_client->delete_segments(make_string_identifier(stream_name),
                                                         make_string_identifier(topic_name), 0, 1),
                 std::exception);
    ASSERT_NO_THROW(unauthenticated_client->connect());
    ASSERT_THROW(unauthenticated_client->delete_segments(make_string_identifier(stream_name),
                                                         make_string_identifier(topic_name), 0, 1),
                 std::exception);
    ASSERT_NO_THROW(unauthenticated_client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(unauthenticated_client->disconnect());
    ASSERT_THROW(unauthenticated_client->delete_segments(make_string_identifier(stream_name),
                                                         make_string_identifier(topic_name), 0, 1),
                 std::exception);
}

TEST_F(LowLevelE2E_Client, DeleteSegmentsOnNonExistentStreamThrows) {
    RecordProperty("description", "Throws when deleting segments from a stream that does not exist.");
    const std::string stream_name         = GetRandomName();
    const std::string topic_name          = GetRandomName();
    const std::string missing_stream_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", "never_expire", 0,
                                         "server_default", {}));

    ASSERT_THROW(
        client->delete_segments(make_string_identifier(missing_stream_name), make_string_identifier(topic_name), 0, 1),
        std::exception);
}

TEST_F(LowLevelE2E_Client, DeleteSegmentsOnNonExistentTopicThrows) {
    RecordProperty("description", "Throws when deleting segments from a topic that does not exist.");
    const std::string stream_name        = GetRandomName();
    const std::string topic_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", "never_expire", 0,
                                         "server_default", {}));

    ASSERT_THROW(
        client->delete_segments(make_string_identifier(stream_name), make_string_identifier(missing_topic_name), 0, 1),
        std::exception);
}

TEST_F(LowLevelE2E_Client, DeleteSegmentsOnNonExistentPartitionThrows) {
    RecordProperty("description", "Throws when deleting segments from a partition that does not exist.");
    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", "never_expire", 0,
                                         "server_default", {}));

    ASSERT_THROW(
        client->delete_segments(make_string_identifier(stream_name), make_string_identifier(topic_name), 999, 1),
        std::exception);
}

TEST_F(LowLevelE2E_Client, DeleteSegmentsWithZeroCountIsNoOp) {
    RecordProperty("description", "Treats delete_segments with count 0 as a no-op.");
    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", "never_expire", 0,
                                         "server_default", {}));

    std::uint32_t stream_id = 0;
    std::uint32_t topic_id  = 0;
    ASSERT_NO_THROW({
        const auto stream_details = client->get_stream(make_string_identifier(stream_name));
        ASSERT_EQ(stream_details.topics.size(), 1u);
        stream_id = stream_details.id;
        topic_id  = stream_details.topics.front().id;
    });

    rust::Vec<iggy::ffi::IggyMessageToSend> messages;
    for (std::uint32_t i = 0; i < 5; ++i) {
        messages.push_back(iggy::ffi::make_message(to_payload("zero-count-" + std::to_string(i)),
                                                   rust::Vec<iggy::ffi::HeaderEntry>()));
    }
    ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id),
                                          "partition_id", partition_id_bytes(0), std::move(messages)));

    iggy::ffi::Partition partition_before_delete{};
    ASSERT_NO_THROW({
        const auto topic_details =
            client->get_topic(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id));
        for (const auto &partition : topic_details.partitions) {
            if (partition.id == 0) {
                partition_before_delete = partition;
                break;
            }
        }
    });

    iggy::ffi::PolledMessages polled_before_delete{};
    ASSERT_NO_THROW({
        polled_before_delete =
            client->poll_messages(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id), 0, "consumer",
                                  make_numeric_identifier(1005), "offset", 0, 1000, false);
    });

    ASSERT_NO_THROW(
        client->delete_segments(make_string_identifier(stream_name), make_string_identifier(topic_name), 0, 0));

    iggy::ffi::Partition partition_after_delete{};
    ASSERT_NO_THROW({
        const auto topic_details =
            client->get_topic(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id));
        for (const auto &partition : topic_details.partitions) {
            if (partition.id == 0) {
                partition_after_delete = partition;
                break;
            }
        }
    });

    iggy::ffi::PolledMessages polled_after_delete{};
    ASSERT_NO_THROW({
        polled_after_delete =
            client->poll_messages(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id), 0, "consumer",
                                  make_numeric_identifier(1006), "offset", 0, 1000, false);
    });

    EXPECT_EQ(partition_after_delete.segments_count, partition_before_delete.segments_count);
    EXPECT_EQ(partition_after_delete.current_offset, partition_before_delete.current_offset);
    EXPECT_EQ(partition_after_delete.messages_count, partition_before_delete.messages_count);
    EXPECT_EQ(partition_after_delete.size_bytes, partition_before_delete.size_bytes);
    EXPECT_EQ(polled_after_delete.count, polled_before_delete.count);
    ASSERT_EQ(polled_after_delete.messages.size(), polled_before_delete.messages.size());
    for (std::size_t i = 0; i < polled_before_delete.messages.size(); ++i) {
        EXPECT_EQ(polled_after_delete.messages[i].offset, polled_before_delete.messages[i].offset);
    }
}

TEST_F(LowLevelE2E_Client, DeleteSegmentsWhenOnlyActiveSegmentRemainsIsNoOp) {
    RecordProperty("description",
                   "Keeps the partition unchanged when delete_segments is called with only the active segment.");
    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", "never_expire", 0,
                                         "server_default", {}));

    std::uint32_t stream_id = 0;
    std::uint32_t topic_id  = 0;
    ASSERT_NO_THROW({
        const auto stream_details = client->get_stream(make_string_identifier(stream_name));
        ASSERT_EQ(stream_details.topics.size(), 1u);
        stream_id = stream_details.id;
        topic_id  = stream_details.topics.front().id;
    });

    rust::Vec<iggy::ffi::IggyMessageToSend> messages;
    for (std::uint32_t i = 0; i < 5; ++i) {
        messages.push_back(iggy::ffi::make_message(to_payload("active-only-" + std::to_string(i)),
                                                   rust::Vec<iggy::ffi::HeaderEntry>()));
    }
    ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id),
                                          "partition_id", partition_id_bytes(0), std::move(messages)));

    iggy::ffi::Partition partition_before_delete{};
    ASSERT_NO_THROW({
        const auto topic_details =
            client->get_topic(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id));
        for (const auto &partition : topic_details.partitions) {
            if (partition.id == 0) {
                partition_before_delete = partition;
                break;
            }
        }
    });
    ASSERT_EQ(partition_before_delete.segments_count, 1u);

    iggy::ffi::PolledMessages polled_before_delete{};
    ASSERT_NO_THROW({
        polled_before_delete =
            client->poll_messages(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id), 0, "consumer",
                                  make_numeric_identifier(1007), "offset", 0, 1000, false);
    });

    ASSERT_NO_THROW(
        client->delete_segments(make_string_identifier(stream_name), make_string_identifier(topic_name), 0, 1));

    iggy::ffi::Partition partition_after_delete{};
    ASSERT_NO_THROW({
        const auto topic_details =
            client->get_topic(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id));
        for (const auto &partition : topic_details.partitions) {
            if (partition.id == 0) {
                partition_after_delete = partition;
                break;
            }
        }
    });

    iggy::ffi::PolledMessages polled_after_delete{};
    ASSERT_NO_THROW({
        polled_after_delete =
            client->poll_messages(make_numeric_identifier(stream_id), make_numeric_identifier(topic_id), 0, "consumer",
                                  make_numeric_identifier(1008), "offset", 0, 1000, false);
    });

    EXPECT_EQ(partition_after_delete.segments_count, partition_before_delete.segments_count);
    EXPECT_EQ(partition_after_delete.current_offset, partition_before_delete.current_offset);
    EXPECT_EQ(partition_after_delete.messages_count, partition_before_delete.messages_count);
    EXPECT_EQ(partition_after_delete.size_bytes, partition_before_delete.size_bytes);
    EXPECT_EQ(polled_after_delete.count, polled_before_delete.count);
    ASSERT_EQ(polled_after_delete.messages.size(), polled_before_delete.messages.size());
    for (std::size_t i = 0; i < polled_before_delete.messages.size(); ++i) {
        EXPECT_EQ(polled_after_delete.messages[i].offset, polled_before_delete.messages[i].offset);
    }
}

// TODO(slbotbm): add a test to create some streams, topics, partitions, and segments, send messages, and create
// consumer groups and verify it.
TEST_F(LowLevelE2E_Client, GetStatsReturnsServerStats) {
    RecordProperty("description",
                   "Returns empty resource counts first, then reflects aggregated streams, topics, partitions, "
                   "consumer groups, and clients.");
    const std::string first_stream_name                 = GetRandomName();
    const std::string second_stream_name                = GetRandomName();
    const std::string first_topic_name                  = GetRandomName();
    const std::string second_topic_name                 = GetRandomName();
    const std::string third_topic_name                  = GetRandomName();
    const std::string first_group_name                  = GetRandomName();
    const std::string second_group_name                 = GetRandomName();
    const std::string third_group_name                  = GetRandomName();
    constexpr std::uint32_t additional_partitions_count = 2;
    iggy::ffi::Client *client                           = GetLoggedInClient();

    iggy::ffi::Client *second_client = nullptr;
    iggy::ffi::Client *third_client  = nullptr;

    iggy::ffi::Stats empty_stats{};
    iggy::ffi::Stats stats_after_create{};
    ASSERT_NO_THROW({
        empty_stats = client->get_stats();
        EXPECT_NE(empty_stats.process_id, 0u);
        EXPECT_GT(empty_stats.threads_count, 0u);
        EXPECT_GT(empty_stats.total_memory, 0u);
        EXPECT_LE(empty_stats.available_memory, empty_stats.total_memory);
        EXPECT_GE(empty_stats.total_disk_space, empty_stats.free_disk_space);
        EXPECT_FALSE(static_cast<std::string>(empty_stats.hostname).empty());
        EXPECT_FALSE(static_cast<std::string>(empty_stats.os_name).empty());
        EXPECT_FALSE(static_cast<std::string>(empty_stats.os_version).empty());
        EXPECT_FALSE(static_cast<std::string>(empty_stats.kernel_version).empty());
        EXPECT_FALSE(static_cast<std::string>(empty_stats.iggy_server_version).empty());
    });

    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_NO_THROW(client->create_topic(make_string_identifier(first_stream_name), first_topic_name, 1, "none",
                                         "server_default", 0, "server_default", {}));
    ASSERT_NO_THROW(client->create_topic(make_string_identifier(first_stream_name), second_topic_name, 2, "none",
                                         "server_default", 0, "server_default", {}));
    ASSERT_NO_THROW(client->create_topic(make_string_identifier(second_stream_name), third_topic_name, 3, "none",
                                         "server_default", 0, "server_default", {}));
    ASSERT_NO_THROW(client->create_partitions(make_string_identifier(first_stream_name),
                                              make_string_identifier(first_topic_name), additional_partitions_count));
    const auto first_group  = client->create_consumer_group(make_string_identifier(first_stream_name),
                                                            make_string_identifier(first_topic_name), first_group_name);
    const auto second_group = client->create_consumer_group(
        make_string_identifier(first_stream_name), make_string_identifier(second_topic_name), second_group_name);
    const auto third_group = client->create_consumer_group(make_string_identifier(second_stream_name),
                                                           make_string_identifier(third_topic_name), third_group_name);

    ASSERT_NO_THROW({ second_client = GetLoggedInClient(); });
    ASSERT_NE(second_client, nullptr);
    ASSERT_NO_THROW({ third_client = GetLoggedInClient(); });
    ASSERT_NE(third_client, nullptr);

    const auto first_stream_details           = client->get_stream(make_string_identifier(first_stream_name));
    const auto second_stream_details          = client->get_stream(make_string_identifier(second_stream_name));
    const std::uint32_t expected_topics_count = first_stream_details.topics_count + second_stream_details.topics_count;
    std::uint32_t first_topic_partitions      = 0;
    std::uint32_t second_topic_partitions     = 0;
    std::uint32_t third_topic_partitions      = 0;
    for (const auto &topic : first_stream_details.topics) {
        if (topic.name == first_topic_name) {
            first_topic_partitions = topic.partitions_count;
        }
        if (topic.name == second_topic_name) {
            second_topic_partitions = topic.partitions_count;
        }
    }
    for (const auto &topic : second_stream_details.topics) {
        if (topic.name == third_topic_name) {
            third_topic_partitions = topic.partitions_count;
        }
    }
    const std::uint32_t expected_partitions_count =
        first_topic_partitions + second_topic_partitions + third_topic_partitions;

    ASSERT_NO_THROW({
        stats_after_create = client->get_stats();
        EXPECT_GE(stats_after_create.streams_count, empty_stats.streams_count + 2u);
        EXPECT_GE(stats_after_create.topics_count, empty_stats.topics_count + expected_topics_count);
        EXPECT_GE(stats_after_create.partitions_count, empty_stats.partitions_count + expected_partitions_count);
        EXPECT_GE(stats_after_create.segments_count, empty_stats.segments_count + expected_partitions_count);
        EXPECT_GE(stats_after_create.consumer_groups_count, empty_stats.consumer_groups_count + 3u);
        EXPECT_GE(stats_after_create.clients_count, empty_stats.clients_count + 2u);
        EXPECT_EQ(first_group.partitions_count, first_topic_partitions);
        EXPECT_EQ(second_group.partitions_count, second_topic_partitions);
        EXPECT_EQ(third_group.partitions_count, third_topic_partitions);
    });

    ASSERT_NO_THROW(client->delete_stream(make_string_identifier(second_stream_name)));
    ForgetTrackedStream(second_stream_name);
    ASSERT_NO_THROW(client->delete_stream(make_string_identifier(first_stream_name)));
    ForgetTrackedStream(first_stream_name);
    DeleteClient(third_client);
    DeleteClient(second_client);

    ASSERT_NO_THROW({
        const auto stats = client->get_stats();
        EXPECT_LE(stats.streams_count, stats_after_create.streams_count);
        EXPECT_LE(stats.topics_count, stats_after_create.topics_count);
        EXPECT_LE(stats.partitions_count, stats_after_create.partitions_count);
        EXPECT_LE(stats.segments_count, stats_after_create.segments_count);
        EXPECT_LE(stats.consumer_groups_count, stats_after_create.consumer_groups_count);
        EXPECT_LE(stats.clients_count, stats_after_create.clients_count);
    });
}

TEST_F(LowLevelE2E_Client, GetStatsIsStableAcrossBackToBackCalls) {
    RecordProperty(
        "description",
        "Returns sane invariant fields across back-to-back get_stats calls on an idle authenticated client.");
    iggy::ffi::Client *client = GetLoggedInClient();

    iggy::ffi::Stats first_stats{};
    iggy::ffi::Stats second_stats{};
    ASSERT_NO_THROW({
        first_stats  = client->get_stats();
        second_stats = client->get_stats();
    });

    EXPECT_NE(first_stats.process_id, 0u);
    EXPECT_NE(second_stats.process_id, 0u);
    EXPECT_EQ(second_stats.process_id, first_stats.process_id);
    EXPECT_GT(first_stats.threads_count, 0u);
    EXPECT_GT(second_stats.threads_count, 0u);
    EXPECT_GT(first_stats.total_memory, 0u);
    EXPECT_GT(second_stats.total_memory, 0u);
    EXPECT_FALSE(static_cast<std::string>(first_stats.hostname).empty());
    EXPECT_FALSE(static_cast<std::string>(second_stats.hostname).empty());
    EXPECT_FALSE(static_cast<std::string>(first_stats.os_name).empty());
    EXPECT_FALSE(static_cast<std::string>(second_stats.os_name).empty());
    EXPECT_FALSE(static_cast<std::string>(first_stats.os_version).empty());
    EXPECT_FALSE(static_cast<std::string>(second_stats.os_version).empty());
    EXPECT_FALSE(static_cast<std::string>(first_stats.kernel_version).empty());
    EXPECT_FALSE(static_cast<std::string>(second_stats.kernel_version).empty());
    EXPECT_FALSE(static_cast<std::string>(first_stats.iggy_server_version).empty());
    EXPECT_FALSE(static_cast<std::string>(second_stats.iggy_server_version).empty());
    EXPECT_EQ(static_cast<std::string>(second_stats.hostname), static_cast<std::string>(first_stats.hostname));
    EXPECT_EQ(static_cast<std::string>(second_stats.os_name), static_cast<std::string>(first_stats.os_name));
    EXPECT_EQ(static_cast<std::string>(second_stats.os_version), static_cast<std::string>(first_stats.os_version));
    EXPECT_EQ(static_cast<std::string>(second_stats.kernel_version),
              static_cast<std::string>(first_stats.kernel_version));
    EXPECT_EQ(static_cast<std::string>(second_stats.iggy_server_version),
              static_cast<std::string>(first_stats.iggy_server_version));
    EXPECT_EQ(second_stats.has_server_semver, first_stats.has_server_semver);
    EXPECT_EQ(second_stats.iggy_server_semver, first_stats.iggy_server_semver);
    EXPECT_GE(first_stats.clients_count, 1u);
    EXPECT_GE(second_stats.clients_count, 1u);
}

TEST_F(LowLevelE2E_Client, GetMeBeforeLoginThrows) {
    RecordProperty("description",
                   "Rejects get_me before connect, after connect but before login, and after disconnect.");
    iggy::ffi::Client *client = GetLoggedOutClient();

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

TEST_F(LowLevelE2E_Client, GetMeReturnsCurrentClientDetails) {
    RecordProperty("description", "Returns the current authenticated client details.");
    iggy::ffi::Client *client = GetLoggedInClient();

    ASSERT_NO_THROW({
        const auto me = client->get_me();
        EXPECT_NE(me.client_id, 0u);
        EXPECT_TRUE(me.has_user_id);
        EXPECT_FALSE(static_cast<std::string>(me.address).empty());
        EXPECT_EQ(static_cast<std::string>(me.transport), "TCP");
        EXPECT_EQ(me.consumer_groups_count, 0u);
        EXPECT_TRUE(me.consumer_groups.empty());
    });
}

TEST_F(LowLevelE2E_Client, GetMeReflectsConsumerGroupMembershipChanges) {
    RecordProperty("description", "Reflects joined consumer groups in get_me and removes them again after leaving.");
    const std::string stream_name = GetRandomName();
    const std::string topic_name  = GetRandomName();
    const std::string group_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);
    const auto created_group = client->create_consumer_group(make_string_identifier(stream_name),
                                                             make_string_identifier(topic_name), group_name);

    std::size_t baseline_groups_size    = 0;
    std::uint32_t baseline_groups_count = 0;
    ASSERT_NO_THROW({
        const auto me         = client->get_me();
        baseline_groups_count = me.consumer_groups_count;
        baseline_groups_size  = me.consumer_groups.size();
    });

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

    ASSERT_NO_THROW({
        const auto me = client->get_me();
        EXPECT_GT(me.consumer_groups_count, baseline_groups_count);
        EXPECT_GT(me.consumer_groups.size(), baseline_groups_size);

        bool found_group = false;
        for (const auto &group : me.consumer_groups) {
            if (group.stream_id != stream_details.id || group.topic_id != stream_details.topics[0].id ||
                group.group_id != created_group.id) {
                continue;
            }
            found_group = true;
            break;
        }
        EXPECT_TRUE(found_group);
    });

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

    ASSERT_NO_THROW({
        const auto me = client->get_me();
        EXPECT_GE(me.consumer_groups_count, baseline_groups_count);
        EXPECT_GE(me.consumer_groups.size(), baseline_groups_size);

        bool found_group = false;
        for (const auto &group : me.consumer_groups) {
            if (group.stream_id != stream_details.id || group.topic_id != stream_details.topics[0].id ||
                group.group_id != created_group.id) {
                continue;
            }
            found_group = true;
            break;
        }
        EXPECT_FALSE(found_group);
    });
}

TEST_F(LowLevelE2E_Client, GetMeIsStableAcrossBackToBackCalls) {
    RecordProperty("description", "Returns stable current-client details across back-to-back get_me calls.");
    iggy::ffi::Client *client = GetLoggedInClient();

    iggy::ffi::ClientInfoDetails first_me{};
    iggy::ffi::ClientInfoDetails second_me{};
    ASSERT_NO_THROW({
        first_me  = client->get_me();
        second_me = client->get_me();
    });

    EXPECT_NE(first_me.client_id, 0u);
    EXPECT_TRUE(first_me.has_user_id);
    EXPECT_TRUE(second_me.has_user_id);
    EXPECT_EQ(second_me.client_id, first_me.client_id);
    EXPECT_EQ(second_me.has_user_id, first_me.has_user_id);
    EXPECT_EQ(second_me.user_id, first_me.user_id);
    EXPECT_EQ(static_cast<std::string>(second_me.address), static_cast<std::string>(first_me.address));
    EXPECT_EQ(static_cast<std::string>(first_me.transport), "TCP");
    EXPECT_EQ(static_cast<std::string>(second_me.transport), "TCP");
    EXPECT_EQ(static_cast<std::string>(second_me.transport), static_cast<std::string>(first_me.transport));
    EXPECT_EQ(second_me.consumer_groups_count, first_me.consumer_groups_count);
    EXPECT_EQ(second_me.consumer_groups.size(), first_me.consumer_groups.size());
}

TEST_F(LowLevelE2E_Client, GetMeReturnsDistinctClientIdsForDifferentSessions) {
    RecordProperty(
        "description",
        "Returns different client ids for separate authenticated sessions while keeping the same user identity.");
    iggy::ffi::Client *first_client  = GetLoggedInClient();
    iggy::ffi::Client *second_client = GetLoggedInClient();

    iggy::ffi::ClientInfoDetails first_me{};
    iggy::ffi::ClientInfoDetails second_me{};
    ASSERT_NO_THROW({
        first_me  = first_client->get_me();
        second_me = second_client->get_me();
    });

    EXPECT_NE(first_me.client_id, 0u);
    EXPECT_NE(second_me.client_id, 0u);
    EXPECT_TRUE(first_me.has_user_id);
    EXPECT_TRUE(second_me.has_user_id);
    EXPECT_NE(second_me.client_id, first_me.client_id);
    EXPECT_EQ(second_me.has_user_id, first_me.has_user_id);
    EXPECT_EQ(second_me.user_id, first_me.user_id);
    EXPECT_EQ(static_cast<std::string>(first_me.transport), "TCP");
    EXPECT_EQ(static_cast<std::string>(second_me.transport), "TCP");
}

TEST_F(LowLevelE2E_Client, GetMeReturnsValidDetailsAfterReconnect) {
    RecordProperty("description",
                   "Returns valid current-client details after reconnecting with a fresh authenticated session.");
    iggy::ffi::Client *first_client = GetLoggedInClient();

    iggy::ffi::ClientInfoDetails first_me{};
    ASSERT_NO_THROW({ first_me = first_client->get_me(); });
    EXPECT_NE(first_me.client_id, 0u);
    EXPECT_TRUE(first_me.has_user_id);

    DeleteClient(first_client);

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

    iggy::ffi::ClientInfoDetails second_me{};
    ASSERT_NO_THROW({ second_me = second_client->get_me(); });
    EXPECT_NE(second_me.client_id, 0u);
    EXPECT_TRUE(second_me.has_user_id);
    EXPECT_EQ(second_me.has_user_id, first_me.has_user_id);
    EXPECT_EQ(second_me.user_id, first_me.user_id);
    EXPECT_EQ(static_cast<std::string>(first_me.transport), "TCP");
    EXPECT_EQ(static_cast<std::string>(second_me.transport), "TCP");
    EXPECT_EQ(static_cast<std::string>(second_me.transport), static_cast<std::string>(first_me.transport));
    EXPECT_FALSE(static_cast<std::string>(second_me.address).empty());
    EXPECT_EQ(second_me.consumer_groups_count, 0u);
    EXPECT_TRUE(second_me.consumer_groups.empty());
}

TEST_F(LowLevelE2E_Client, GetClientBeforeLoginThrows) {
    RecordProperty("description",
                   "Rejects get_client before connect, after connect but before login, and after disconnect.");
    iggy::ffi::Client *client = GetLoggedOutClient();

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

TEST_F(LowLevelE2E_Client, GetClientWithWrongClientIdThrows) {
    RecordProperty("description", "Rejects querying invalid or non-existent client ids.");
    iggy::ffi::Client *client = GetLoggedInClient();

    std::uint32_t non_existent_client_id = 1u;
    ASSERT_NO_THROW({
        const auto clients = client->get_clients();
        std::unordered_set<std::uint32_t> client_ids;
        for (const auto &entry : clients) {
            client_ids.insert(entry.client_id);
        }

        while (client_ids.find(non_existent_client_id) != client_ids.end()) {
            ++non_existent_client_id;
        }
    });

    const std::uint32_t wrong_client_ids[] = {0u, non_existent_client_id};
    for (const std::uint32_t wrong_client_id : wrong_client_ids) {
        SCOPED_TRACE(wrong_client_id);
        ASSERT_THROW(client->get_client(wrong_client_id), std::exception);
    }
}

TEST_F(LowLevelE2E_Client, GetClientReturnsDetailsForMatchingClientId) {
    RecordProperty("description", "Returns current client details when querying with the authenticated client id.");
    iggy::ffi::Client *client = GetLoggedInClient();

    iggy::ffi::ClientInfoDetails current_client{};
    iggy::ffi::ClientInfoDetails looked_up_client{};
    ASSERT_NO_THROW({
        current_client   = client->get_me();
        looked_up_client = client->get_client(current_client.client_id);
    });

    EXPECT_NE(current_client.client_id, 0u);
    EXPECT_TRUE(current_client.has_user_id);
    EXPECT_TRUE(looked_up_client.has_user_id);
    EXPECT_EQ(looked_up_client.client_id, current_client.client_id);
    EXPECT_EQ(looked_up_client.has_user_id, current_client.has_user_id);
    EXPECT_EQ(looked_up_client.user_id, current_client.user_id);
    EXPECT_EQ(static_cast<std::string>(looked_up_client.address), static_cast<std::string>(current_client.address));
    EXPECT_EQ(static_cast<std::string>(looked_up_client.transport), "TCP");
    EXPECT_EQ(static_cast<std::string>(looked_up_client.transport), static_cast<std::string>(current_client.transport));
    EXPECT_EQ(looked_up_client.consumer_groups_count, current_client.consumer_groups_count);
    EXPECT_EQ(looked_up_client.consumer_groups.size(), current_client.consumer_groups.size());
}

TEST_F(LowLevelE2E_Client, GetClientIsStableAcrossBackToBackCalls) {
    RecordProperty("description", "Returns stable client details across back-to-back get_client calls.");
    iggy::ffi::Client *client = GetLoggedInClient();

    iggy::ffi::ClientInfoDetails current_client{};
    iggy::ffi::ClientInfoDetails first_lookup{};
    iggy::ffi::ClientInfoDetails second_lookup{};
    ASSERT_NO_THROW({
        current_client = client->get_me();
        first_lookup   = client->get_client(current_client.client_id);
        second_lookup  = client->get_client(current_client.client_id);
    });

    EXPECT_NE(current_client.client_id, 0u);
    EXPECT_TRUE(current_client.has_user_id);
    EXPECT_TRUE(first_lookup.has_user_id);
    EXPECT_TRUE(second_lookup.has_user_id);
    EXPECT_EQ(first_lookup.client_id, current_client.client_id);
    EXPECT_EQ(second_lookup.client_id, first_lookup.client_id);
    EXPECT_EQ(first_lookup.has_user_id, current_client.has_user_id);
    EXPECT_EQ(second_lookup.has_user_id, first_lookup.has_user_id);
    EXPECT_EQ(second_lookup.user_id, first_lookup.user_id);
    EXPECT_EQ(static_cast<std::string>(second_lookup.address), static_cast<std::string>(first_lookup.address));
    EXPECT_EQ(static_cast<std::string>(first_lookup.transport), "TCP");
    EXPECT_EQ(static_cast<std::string>(second_lookup.transport), "TCP");
    EXPECT_EQ(static_cast<std::string>(second_lookup.transport), static_cast<std::string>(first_lookup.transport));
    EXPECT_EQ(second_lookup.consumer_groups_count, first_lookup.consumer_groups_count);
    EXPECT_EQ(second_lookup.consumer_groups.size(), first_lookup.consumer_groups.size());
}

TEST_F(LowLevelE2E_Client, GetClientsBeforeLoginThrows) {
    RecordProperty("description",
                   "Rejects get_clients before connect, after connect but before login, and after disconnect.");
    iggy::ffi::Client *client = GetLoggedOutClient();

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

TEST_F(LowLevelE2E_Client, GetClientsReturnsActiveClientSessions) {
    RecordProperty("description", "Returns the currently active authenticated client sessions.");
    iggy::ffi::Client *first_client  = GetLoggedInClient();
    iggy::ffi::Client *second_client = GetLoggedInClient();

    iggy::ffi::ClientInfoDetails first_me{};
    iggy::ffi::ClientInfoDetails second_me{};
    rust::Vec<iggy::ffi::ClientInfo> clients;
    ASSERT_NO_THROW({
        first_me  = first_client->get_me();
        second_me = second_client->get_me();
        clients   = first_client->get_clients();
    });

    ASSERT_GE(clients.size(), 2u);

    bool found_first  = false;
    bool found_second = false;
    for (const auto &client : clients) {
        EXPECT_NE(client.client_id, 0u);
        EXPECT_EQ(static_cast<std::string>(client.transport), "TCP");

        if (client.client_id == first_me.client_id) {
            found_first = true;
            EXPECT_EQ(client.has_user_id, first_me.has_user_id);
            EXPECT_EQ(client.user_id, first_me.user_id);
            EXPECT_EQ(static_cast<std::string>(client.address), static_cast<std::string>(first_me.address));
            EXPECT_EQ(client.consumer_groups_count, first_me.consumer_groups_count);
        }

        if (client.client_id == second_me.client_id) {
            found_second = true;
            EXPECT_EQ(client.has_user_id, second_me.has_user_id);
            EXPECT_EQ(client.user_id, second_me.user_id);
            EXPECT_EQ(static_cast<std::string>(client.address), static_cast<std::string>(second_me.address));
            EXPECT_EQ(client.consumer_groups_count, second_me.consumer_groups_count);
        }
    }

    EXPECT_TRUE(found_first);
    EXPECT_TRUE(found_second);
}

TEST_F(LowLevelE2E_Client, GetClientsIsStableAcrossBackToBackCalls) {
    RecordProperty("description", "Returns stable client lists across back-to-back get_clients calls.");
    iggy::ffi::Client *first_client  = GetLoggedInClient();
    iggy::ffi::Client *second_client = GetLoggedInClient();
    ASSERT_NE(second_client, nullptr);

    iggy::ffi::ClientInfoDetails first_me{};
    iggy::ffi::ClientInfoDetails second_me{};
    rust::Vec<iggy::ffi::ClientInfo> first_clients;
    rust::Vec<iggy::ffi::ClientInfo> second_clients;
    ASSERT_NO_THROW({
        first_me       = first_client->get_me();
        second_me      = second_client->get_me();
        first_clients  = first_client->get_clients();
        second_clients = first_client->get_clients();
    });

    ASSERT_GE(first_clients.size(), 2u);
    ASSERT_GE(second_clients.size(), 2u);

    const auto expect_entry_matches = [](const rust::Vec<iggy::ffi::ClientInfo> &clients,
                                         const iggy::ffi::ClientInfoDetails &expected) {
        bool found = false;
        for (const auto &entry : clients) {
            if (entry.client_id != expected.client_id) {
                continue;
            }

            found = true;
            EXPECT_EQ(entry.has_user_id, expected.has_user_id);
            EXPECT_EQ(entry.user_id, expected.user_id);
            EXPECT_EQ(static_cast<std::string>(entry.address), static_cast<std::string>(expected.address));
            EXPECT_EQ(static_cast<std::string>(entry.transport), static_cast<std::string>(expected.transport));
            EXPECT_EQ(entry.consumer_groups_count, expected.consumer_groups_count);
            break;
        }

        EXPECT_TRUE(found);
    };

    expect_entry_matches(first_clients, first_me);
    expect_entry_matches(first_clients, second_me);
    expect_entry_matches(second_clients, first_me);
    expect_entry_matches(second_clients, second_me);
}

TEST_F(LowLevelE2E_Client, GetClientsMatchesGetClientForReturnedIds) {
    RecordProperty("description", "Returns list entries that agree with get_client for each returned client id.");
    iggy::ffi::Client *first_client  = GetLoggedInClient();
    iggy::ffi::Client *second_client = GetLoggedInClient();
    ASSERT_NE(second_client, nullptr);

    rust::Vec<iggy::ffi::ClientInfo> clients;
    ASSERT_NO_THROW({ clients = first_client->get_clients(); });
    ASSERT_GE(clients.size(), 2u);

    for (const auto &client : clients) {
        SCOPED_TRACE(client.client_id);
        iggy::ffi::ClientInfoDetails details{};
        ASSERT_NO_THROW({ details = first_client->get_client(client.client_id); });

        EXPECT_EQ(details.client_id, client.client_id);
        EXPECT_EQ(details.has_user_id, client.has_user_id);
        EXPECT_EQ(details.user_id, client.user_id);
        EXPECT_EQ(static_cast<std::string>(details.address), static_cast<std::string>(client.address));
        EXPECT_EQ(static_cast<std::string>(details.transport), static_cast<std::string>(client.transport));
        EXPECT_EQ(details.consumer_groups_count, client.consumer_groups_count);
    }
}

TEST_F(LowLevelE2E_Client, GetClientsReflectsAdditionalSession) {
    RecordProperty("description", "Reflects a newly added authenticated session in subsequent get_clients results.");
    iggy::ffi::Client *first_client = GetLoggedInClient();

    rust::Vec<iggy::ffi::ClientInfo> clients_before;
    ASSERT_NO_THROW({ clients_before = first_client->get_clients(); });

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

    iggy::ffi::ClientInfoDetails second_me{};
    rust::Vec<iggy::ffi::ClientInfo> clients_after;
    ASSERT_NO_THROW({
        second_me     = second_client->get_me();
        clients_after = first_client->get_clients();
    });

    bool found_before = false;
    for (const auto &client : clients_before) {
        if (client.client_id == second_me.client_id) {
            found_before = true;
            break;
        }
    }
    EXPECT_FALSE(found_before);

    bool found_after = false;
    for (const auto &client : clients_after) {
        if (client.client_id != second_me.client_id) {
            continue;
        }

        found_after = true;
        EXPECT_EQ(client.has_user_id, second_me.has_user_id);
        EXPECT_EQ(client.user_id, second_me.user_id);
        EXPECT_EQ(static_cast<std::string>(client.address), static_cast<std::string>(second_me.address));
        EXPECT_EQ(static_cast<std::string>(client.transport), "TCP");
        EXPECT_EQ(client.consumer_groups_count, second_me.consumer_groups_count);
        break;
    }
    EXPECT_TRUE(found_after);
}

TEST_F(LowLevelE2E_Client, GetClusterMetadataBeforeLoginThrows) {
    RecordProperty("description",
                   "Rejects get_cluster_metadata before connect, after connect but before login, and after disconnect, "
                   "and serves it once authenticated.");
    iggy::ffi::Client *client = GetLoggedOutClient();

    ASSERT_THROW(client->get_cluster_metadata(), std::exception);
    ASSERT_NO_THROW(client->connect());
    // The roster is private, so the read is auth-gated. No pre-login read is
    // needed: a client that dialed a backup logs in there and the server
    // forwards the register to the primary.
    ASSERT_THROW(client->get_cluster_metadata(), std::exception);
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    iggy::ffi::ClusterMetadata metadata{};
    ASSERT_NO_THROW({ metadata = client->get_cluster_metadata(); });
    ASSERT_GE(metadata.nodes.size(), 1u);
    ASSERT_NO_THROW(client->disconnect());
    ASSERT_THROW(client->get_cluster_metadata(), std::exception);
}

TEST_F(LowLevelE2E_Client, GetClusterMetadataReturnsSingleNodeMetadata) {
    RecordProperty("description",
                   "Returns the expected single-node cluster metadata shape from the default test server.");
    iggy::ffi::Client *client = GetLoggedInClient();

    iggy::ffi::ClusterMetadata metadata{};
    ASSERT_NO_THROW({ metadata = client->get_cluster_metadata(); });

    EXPECT_EQ(static_cast<std::string>(metadata.name), "single-node");
    ASSERT_EQ(metadata.nodes.size(), 1u);

    const auto &node = metadata.nodes[0];
    EXPECT_FALSE(static_cast<std::string>(node.name).empty());
    EXPECT_FALSE(static_cast<std::string>(node.ip).empty());
    EXPECT_EQ(static_cast<std::string>(node.role), "leader");
    EXPECT_EQ(static_cast<std::string>(node.status), "healthy");
    EXPECT_NE(node.endpoints.tcp, 0u);
    EXPECT_NE(node.endpoints.http, 0u);
}

TEST_F(LowLevelE2E_Client, GetClusterMetadataIsStableAcrossBackToBackCalls) {
    RecordProperty("description", "Returns stable single-node cluster metadata across back-to-back calls.");
    iggy::ffi::Client *client = GetLoggedInClient();

    iggy::ffi::ClusterMetadata first_metadata{};
    iggy::ffi::ClusterMetadata second_metadata{};
    ASSERT_NO_THROW({
        first_metadata  = client->get_cluster_metadata();
        second_metadata = client->get_cluster_metadata();
    });

    EXPECT_EQ(static_cast<std::string>(first_metadata.name), static_cast<std::string>(second_metadata.name));
    ASSERT_EQ(first_metadata.nodes.size(), 1u);
    ASSERT_EQ(second_metadata.nodes.size(), 1u);

    const auto &first_node  = first_metadata.nodes[0];
    const auto &second_node = second_metadata.nodes[0];
    EXPECT_EQ(static_cast<std::string>(first_node.name), static_cast<std::string>(second_node.name));
    EXPECT_EQ(static_cast<std::string>(first_node.ip), static_cast<std::string>(second_node.ip));
    EXPECT_EQ(static_cast<std::string>(first_node.role), static_cast<std::string>(second_node.role));
    EXPECT_EQ(static_cast<std::string>(first_node.status), static_cast<std::string>(second_node.status));
    EXPECT_EQ(first_node.endpoints.tcp, second_node.endpoints.tcp);
    EXPECT_EQ(first_node.endpoints.quic, second_node.endpoints.quic);
    EXPECT_EQ(first_node.endpoints.http, second_node.endpoints.http);
    EXPECT_EQ(first_node.endpoints.websocket, second_node.endpoints.websocket);
}

TEST_F(LowLevelE2E_Client, PingSucceedsForNewConnection) {
    RecordProperty("description", "Successfully pings the server from a fresh unauthenticated client session.");
    iggy::ffi::Client *client = GetLoggedOutClient();

    // The VSR client has no lazy connect; ping still needs no authentication.
    ASSERT_NO_THROW(client->connect());
    ASSERT_NO_THROW(client->ping());
}

TEST_F(LowLevelE2E_Client, HeartbeatIntervalReturnsDefaultValueForNewConnection) {
    RecordProperty("description",
                   "Returns the default heartbeat interval in microseconds for a fresh unauthenticated client.");
    constexpr std::uint64_t default_heartbeat_micros = 5'000'000ull;
    iggy::ffi::Client *client                        = GetLoggedOutClient();

    const auto heartbeat_interval = client->heartbeat_interval();
    EXPECT_EQ(heartbeat_interval, default_heartbeat_micros);
}

TEST_F(LowLevelE2E_Client, HeartbeatIntervalReturnsConfiguredValueFromConnectionString) {
    RecordProperty("description",
                   "Returns the configured heartbeat interval in microseconds from the connection string.");
    constexpr std::uint64_t configured_heartbeat_micros = 10'000'000ull;
    iggy::ffi::Client *client                           = nullptr;
    ASSERT_NO_THROW(
        { client = iggy::ffi::from_connection_string("iggy://iggy:iggy@127.0.0.1:8090?heartbeat_interval=10s"); });
    ASSERT_NE(client, nullptr);
    TrackClient(client);

    const auto heartbeat_interval = client->heartbeat_interval();
    EXPECT_EQ(heartbeat_interval, configured_heartbeat_micros);
}

TEST_F(LowLevelE2E_Client, SnapshotBeforeLoginThrows) {
    RecordProperty("description",
                   "Rejects snapshot before connect, after connect but before login, and after disconnect.");
    iggy::ffi::Client *client = GetLoggedOutClient();

    ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"test"})), std::exception);

    ASSERT_NO_THROW(client->connect());
    ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"test"})), std::exception);
    ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
    ASSERT_NO_THROW(client->disconnect());
    ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"test"})), std::exception);
}

TEST_F(LowLevelE2E_Client, SnapshotAllCombinedWithOtherTypeThrows) {
    RecordProperty("description", "Rejects combining the all snapshot type with any other snapshot type.");
    iggy::ffi::Client *client = GetLoggedInClient();

    ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"all", "test"})), std::exception);
}

TEST_F(LowLevelE2E_Client, SnapshotWithEmptySnapshotTypesThrows) {
    RecordProperty("description", "Rejects an empty snapshot type list in the wrapper before sending.");
    iggy::ffi::Client *client = GetLoggedInClient();

    rust::Vec<rust::String> snapshot_types;
    ASSERT_THROW(client->snapshot("deflated", snapshot_types), std::exception);
}

TEST_F(LowLevelE2E_Client, SnapshotReturnsNonEmptyBytes) {
    RecordProperty("description", "Returns a non-empty snapshot for a valid compression and snapshot type.");
    iggy::ffi::Client *client = GetLoggedInClient();

    rust::Vec<std::uint8_t> snapshot_bytes;
    ASSERT_NO_THROW({ snapshot_bytes = client->snapshot("deflated", make_snapshot_types({"test"})); });
    EXPECT_FALSE(snapshot_bytes.empty());
}

TEST_F(LowLevelE2E_Client, SnapshotWithInvalidCompressionThrows) {
    RecordProperty("description",
                   "Rejects empty or invalid snapshot compression values in the wrapper before sending.");
    iggy::ffi::Client *client = GetLoggedInClient();

    ASSERT_THROW(client->snapshot("", make_snapshot_types({"test"})), std::exception);
    ASSERT_THROW(client->snapshot("invalid-compression", make_snapshot_types({"test"})), std::exception);
}

TEST_F(LowLevelE2E_Client, SnapshotWithInvalidSnapshotTypeThrows) {
    RecordProperty("description", "Rejects invalid snapshot type values in the wrapper before sending.");
    iggy::ffi::Client *client = GetLoggedInClient();

    ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"not-a-real-type"})), std::exception);
}

TEST_F(LowLevelE2E_Client, SendBinaryRequestPingReturnsEmptyBytes) {
    RecordProperty("description", "Returns an empty response body for a raw ping command with an empty payload.");
    constexpr std::uint32_t ping_command_code = 1;
    iggy::ffi::Client *client                 = GetLoggedInClient();

    rust::Vec<std::uint8_t> empty_payload;
    rust::Vec<std::uint8_t> response;
    ASSERT_NO_THROW({ response = client->send_binary_request(ping_command_code, empty_payload); });
    EXPECT_TRUE(response.empty());
}

TEST_F(LowLevelE2E_Client, SendBinaryRequestGetStatsReturnsNonEmptyBytes) {
    RecordProperty("description",
                   "Returns a non-empty response body for a raw get-stats command with an empty payload.");
    constexpr std::uint32_t get_stats_command_code = 10;
    iggy::ffi::Client *client                      = GetLoggedInClient();

    rust::Vec<std::uint8_t> empty_payload;
    rust::Vec<std::uint8_t> response;
    ASSERT_NO_THROW({ response = client->send_binary_request(get_stats_command_code, empty_payload); });
    EXPECT_FALSE(response.empty());
}

TEST_F(LowLevelE2E_Client, SendBinaryRequestLoginUserCodeThrows) {
    RecordProperty("description",
                   "Rejects the login-user session-control code client-side before it reaches the server.");
    constexpr std::uint32_t login_user_command_code = 38;
    iggy::ffi::Client *client                       = GetLoggedInClient();

    rust::Vec<std::uint8_t> empty_payload;
    ASSERT_THROW(client->send_binary_request(login_user_command_code, empty_payload), std::exception);
}

TEST_F(LowLevelE2E_Client, SendBinaryRequestUnknownCommandCodeThrows) {
    RecordProperty("description", "Rejects an unknown command code with an invalid-command error from the server.");
    constexpr std::uint32_t unknown_command_code = 60000;
    iggy::ffi::Client *client                    = GetLoggedInClient();

    rust::Vec<std::uint8_t> empty_payload;
    ASSERT_THROW(client->send_binary_request(unknown_command_code, empty_payload), std::exception);
}