#pragma once

#include "universal_container/adaptation_policy.hpp"
#include "universal_container/flat_hash_index.hpp"
#include "universal_container/tiered_storage.hpp"

#include <algorithm>
#include <cassert>
#include <compare>
#include <concepts>
#include <cstddef>
#include <cstdint>
#include <cmath>
#include <functional>
#include <iterator>
#include <limits>
#include <optional>
#include <span>
#include <stdexcept>
#include <type_traits>
#include <utility>
#include <variant>
#include <vector>

namespace uc {

template <class T,
          bool HashIndexEnabled = false,
          class AdaptationPolicy = ResizePolicy,
          class Hash = std::hash<T>,
          class Equal = std::equal_to<T>>
class AdaptiveSequence {
public:
    using value_type = T;
    using size_type = std::size_t;
    using difference_type = std::ptrdiff_t;
    using stable_id = std::uint64_t;
    static constexpr bool hash_index_enabled = HashIndexEnabled;
    static constexpr stable_id invalid_id = std::numeric_limits<stable_id>::max();

private:
    using internal_id = typename detail::FlatDuplicateIndex<T, Hash, Equal>::id_type;
    static constexpr internal_id invalid_internal_id =
        detail::FlatDuplicateIndex<T, Hash, Equal>::invalid_id;

    struct IndexedRecord {
        T value;
        internal_id id = invalid_internal_id;
    };

    using record_type = std::conditional_t<HashIndexEnabled, IndexedRecord, T>;
    using vector_storage = std::vector<record_type>;
    using tiered_storage = detail::TieredStorage<record_type>;

    struct IdMetadata {
        internal_id previous = invalid_internal_id;
        internal_id next = invalid_internal_id;
        internal_id free_next = invalid_internal_id;
        std::uint32_t primary = 0;   // vector index or stable leaf id
        std::uint32_t secondary = 0; // local leaf index
        std::uint32_t generation = 0;
        std::uint8_t flags = 0;
    };

    static constexpr std::uint8_t alive_flag = 0x01;
    static constexpr std::uint8_t tiered_flag = 0x02;

    struct IndexedState {
        detail::FlatDuplicateIndex<T, Hash, Equal> values;
        std::vector<IdMetadata> ids;
        internal_id free_head = invalid_internal_id;
    };

    struct NoIndexState {};
    using index_state_type = std::conditional_t<HashIndexEnabled, IndexedState, NoIndexState>;

public:
    class reference_proxy {
    public:
        reference_proxy(AdaptiveSequence& owner, size_type index) noexcept
            : owner_(&owner), index_(index), generation_(owner.generation_) {}

        reference_proxy& operator=(const T& value) {
            validate();
            owner_->set(index_, value);
            return *this;
        }

        reference_proxy& operator=(T&& value) {
            validate();
            owner_->set(index_, std::move(value));
            return *this;
        }

        reference_proxy& operator=(const reference_proxy& other) {
            return *this = static_cast<const T&>(other);
        }

        operator const T&() const {
            validate();
            return owner_->value_at_unchecked(index_);
        }
        const T* operator->() const {
            validate();
            return &owner_->value_at_unchecked(index_);
        }
        const T& get() const {
            validate();
            return owner_->value_at_unchecked(index_);
        }

    private:
        void validate() const {
            if (generation_ != owner_->generation_) {
                throw std::logic_error("AdaptiveSequence reference proxy was invalidated");
            }
        }

        AdaptiveSequence* owner_;
        size_type index_;
        std::uint64_t generation_;
    };

    using reference = std::conditional_t<HashIndexEnabled, reference_proxy, T&>;
    using const_reference = const T&;

    template <bool Const>
    class basic_iterator {
        friend class AdaptiveSequence;
        template <bool>
        friend class basic_iterator;

        using owner_type = std::conditional_t<Const, const AdaptiveSequence, AdaptiveSequence>;

        basic_iterator(owner_type* owner, size_type index) noexcept
            : owner_(owner), index_(index), generation_(owner ? owner->generation_ : 0) {}

        void validate() const {
            if (owner_ && generation_ != owner_->generation_) {
                throw std::logic_error("AdaptiveSequence iterator was invalidated");
            }
        }

    public:
        using iterator_category = std::random_access_iterator_tag;
        using iterator_concept = std::random_access_iterator_tag;
        using value_type = T;
        using difference_type = std::ptrdiff_t;
        using reference = std::conditional_t<Const, const T&,
            typename AdaptiveSequence::reference>;
        using pointer = const T*;

        basic_iterator() = default;

        template <bool OtherConst>
            requires(Const && !OtherConst)
        basic_iterator(const basic_iterator<OtherConst>& other) noexcept
            : owner_(other.owner_), index_(other.index_), generation_(other.generation_) {}

        reference operator*() const {
            validate();
            if constexpr (Const) {
                return owner_->value_at_unchecked(index_);
            } else if constexpr (HashIndexEnabled) {
                return reference(*owner_, index_);
            } else {
                return owner_->value_at_unchecked(index_);
            }
        }

        pointer operator->() const {
            validate();
            return &owner_->value_at_unchecked(index_);
        }

        reference operator[](difference_type offset) const { return *(*this + offset); }

        basic_iterator& operator++() noexcept { ++index_; return *this; }
        basic_iterator operator++(int) noexcept { auto copy = *this; ++*this; return copy; }
        basic_iterator& operator--() noexcept { --index_; return *this; }
        basic_iterator operator--(int) noexcept { auto copy = *this; --*this; return copy; }
        basic_iterator& operator+=(difference_type offset) noexcept {
            index_ = static_cast<size_type>(static_cast<difference_type>(index_) + offset);
            return *this;
        }
        basic_iterator& operator-=(difference_type offset) noexcept { return *this += -offset; }

        friend basic_iterator operator+(basic_iterator iterator, difference_type offset) noexcept {
            iterator += offset;
            return iterator;
        }
        friend basic_iterator operator+(difference_type offset, basic_iterator iterator) noexcept {
            iterator += offset;
            return iterator;
        }
        friend basic_iterator operator-(basic_iterator iterator, difference_type offset) noexcept {
            iterator -= offset;
            return iterator;
        }
        friend difference_type operator-(const basic_iterator& left,
                                         const basic_iterator& right) noexcept {
            return static_cast<difference_type>(left.index_)
                 - static_cast<difference_type>(right.index_);
        }
        friend bool operator==(const basic_iterator&, const basic_iterator&) = default;
        friend auto operator<=>(const basic_iterator& left,
                                const basic_iterator& right) noexcept {
            return left.index_ <=> right.index_;
        }

    private:
        owner_type* owner_ = nullptr;
        size_type index_ = 0;
        std::uint64_t generation_ = 0;
    };

    using iterator = basic_iterator<false>;
    using const_iterator = basic_iterator<true>;

    explicit AdaptiveSequence(TieredConfig tiered_config = {},
                              AdaptationPolicy policy = AdaptationPolicy{})
        : tiered_config_(tiered_config), policy_(std::move(policy)), storage_(vector_storage{}) {
        if constexpr (requires(AdaptationPolicy& p) { p.set_tiered_config(tiered_config_); }) {
            policy_.set_tiered_config(tiered_config_);
        }
        read_sample_rate_ = configured_read_sample_rate();
        read_sample_countdown_ = read_sample_rate_;
        edit_sample_rate_ = configured_edit_sample_rate();
        edit_sample_countdown_ = edit_sample_rate_;
    }

    template <std::input_iterator InputIt, std::sentinel_for<InputIt> Sentinel>
    AdaptiveSequence(InputIt first,
                     Sentinel last,
                     TieredConfig tiered_config = {},
                     AdaptationPolicy policy = AdaptationPolicy{})
        : AdaptiveSequence(tiered_config, std::move(policy)) {
        for (; first != last; ++first) {
            push_back(*first);
        }
    }

    AdaptiveSequence(const AdaptiveSequence&) = default;

    AdaptiveSequence(AdaptiveSequence&& other) noexcept
        : tiered_config_(std::move(other.tiered_config_)),
          policy_(std::move(other.policy_)),
          storage_(std::move(other.storage_)),
          index_state_(std::move(other.index_state_)),
          equal_(std::move(other.equal_)),
          residency_(other.residency_),
          read_adaptation_(other.read_adaptation_),
          logical_capacity_(other.logical_capacity_),
          read_sample_rate_(other.read_sample_rate_),
          read_sample_countdown_(other.read_sample_countdown_),
          edit_sample_rate_(other.edit_sample_rate_),
          edit_sample_countdown_(other.edit_sample_countdown_),
          generation_(other.generation_) {
        other.reset_after_move();
    }

    AdaptiveSequence& operator=(const AdaptiveSequence& other) {
        if (this != &other) {
            AdaptiveSequence replacement(other);
            *this = std::move(replacement);
        }
        return *this;
    }

    AdaptiveSequence& operator=(AdaptiveSequence&& other) noexcept {
        if (this == &other) {
            return *this;
        }
        const auto invalidated_generation = generation_ + 1;
        tiered_config_ = std::move(other.tiered_config_);
        policy_ = std::move(other.policy_);
        storage_ = std::move(other.storage_);
        index_state_ = std::move(other.index_state_);
        equal_ = std::move(other.equal_);
        residency_ = other.residency_;
        read_adaptation_ = other.read_adaptation_;
        logical_capacity_ = other.logical_capacity_;
        read_sample_rate_ = other.read_sample_rate_;
        read_sample_countdown_ = other.read_sample_countdown_;
        edit_sample_rate_ = other.edit_sample_rate_;
        edit_sample_countdown_ = other.edit_sample_countdown_;
        generation_ = invalidated_generation;
        other.reset_after_move();
        return *this;
    }
    ~AdaptiveSequence() = default;

    [[nodiscard]] size_type size() const noexcept {
        return mode() == StorageMode::vector
            ? std::get<vector_storage>(storage_).size()
            : std::get<tiered_storage>(storage_).size();
    }

    [[nodiscard]] bool empty() const noexcept { return size() == 0; }
    // Logical capacity is shared by both backends.  It is the only trigger for
    // automatic representation/geometry changes; TieredStorage's internal
    // leaf slack is intentionally not exposed as container capacity.
    [[nodiscard]] size_type capacity() const noexcept { return logical_capacity_; }
    [[nodiscard]] StorageMode mode() const noexcept {
        return std::holds_alternative<vector_storage>(storage_)
            ? StorageMode::vector
            : StorageMode::tiered;
    }
    [[nodiscard]] ResidencyMode residency_mode() const noexcept { return residency_; }
    [[nodiscard]] const TieredConfig& tiered_config() const noexcept { return tiered_config_; }

    reference operator[](size_type index) {
        observe_random_read();
        if constexpr (HashIndexEnabled) {
            return reference_proxy(*this, index);
        } else {
            return value_at_unchecked(index);
        }
    }

    const_reference operator[](size_type index) const noexcept {
        observe_random_read();
        return value_at_unchecked(index);
    }

    reference at(size_type index) {
        check_index(index);
        return (*this)[index];
    }

    const_reference at(size_type index) const {
        check_index(index);
        return (*this)[index];
    }

    void push_back(const T& value) { push_back_impl(T(value)); }
    void push_back(T&& value) { push_back_impl(std::move(value)); }

    void insert(size_type index, const T& value) { insert_impl(index, T(value)); }
    void insert(size_type index, T&& value) { insert_impl(index, std::move(value)); }

    void erase(size_type index) {
        check_index(index);
        const auto old_size = size();
        const auto id = record_id(record_at_unchecked(index));
        if constexpr (HashIndexEnabled) {
            unlink_value(value_at_unchecked(index), id);
        }

        auto relocate = relocation_callback();
        if (mode() == StorageMode::vector) {
            auto& values = std::get<vector_storage>(storage_);
            values.erase(values.begin() + static_cast<difference_type>(index));
            refresh_vector_locations(index);
        } else {
            (void)std::get<tiered_storage>(storage_).erase(index, relocate);
        }
        if constexpr (HashIndexEnabled) {
            release_id_slot(id);
        }
        ++generation_;
        observe_structural_edit(OperationKind::erase, old_size, index);
        shrink_after_erase();
    }

    void set(size_type index, const T& value) { set_impl(index, T(value)); }
    void set(size_type index, T&& value) { set_impl(index, std::move(value)); }

    void clear() {
        if constexpr (HashIndexEnabled) {
            index_state_.values.clear();
            rebuild_free_id_list();
        }
        storage_.template emplace<vector_storage>();
        logical_capacity_ = 0;
        residency_ = ResidencyMode::automatic;
        policy_.reset();
        read_sample_countdown_ = read_sample_rate_;
        edit_sample_countdown_ = edit_sample_rate_;
        ++generation_;
    }

    void reserve(size_type capacity) {
        force_vector_mode();
        const auto requested = std::max(capacity, size());
        std::get<vector_storage>(storage_).reserve(requested);
        logical_capacity_ = std::max(logical_capacity_, requested);
        if constexpr (HashIndexEnabled) {
            index_state_.values.reserve_for_elements(logical_capacity_);
            const auto additional = requested > size() ? requested - size() : 0;
            index_state_.ids.reserve(index_state_.ids.size() + additional);
        }
        ++generation_;
    }

    [[nodiscard]] bool contains(const T& value) const {
        if constexpr (HashIndexEnabled) {
            return index_state_.values.find(value) != nullptr;
        } else {
            return find_one(value).has_value();
        }
    }

    [[nodiscard]] std::optional<size_type> find_one(const T& value) const {
        if constexpr (HashIndexEnabled) {
            const auto* entry = index_state_.values.find(value);
            if (!entry || entry->head == invalid_internal_id) {
                return std::nullopt;
            }
            return resolve_internal_id(entry->head);
        } else {
            if (mode() == StorageMode::vector) {
                const auto& values = std::get<vector_storage>(storage_);
                for (size_type i = 0; i < values.size(); ++i) {
                    if (equal_(record_value(values[i]), value)) {
                        return i;
                    }
                }
                return std::nullopt;
            }
            const auto found = std::get<tiered_storage>(storage_).find_if(
                [&](const record_type& record) {
                    return equal_(record_value(record), value);
                });
            return found == tiered_storage::npos
                ? std::nullopt : std::optional<size_type>(found);
        }
    }

    [[nodiscard]] std::vector<size_type> find_all(const T& value) const {
        std::vector<size_type> result;
        if constexpr (HashIndexEnabled) {
            const auto* entry = index_state_.values.find(value);
            if (!entry) {
                return result;
            }
            result.reserve(entry->count);
            auto id = entry->head;
            while (id != invalid_internal_id) {
                result.push_back(resolve_internal_id(id));
                id = index_state_.ids[static_cast<size_type>(id)].next;
            }
            std::sort(result.begin(), result.end());
        } else {
            if (mode() == StorageMode::vector) {
                const auto& values = std::get<vector_storage>(storage_);
                for (size_type i = 0; i < values.size(); ++i) {
                    if (equal_(record_value(values[i]), value)) {
                        result.push_back(i);
                    }
                }
            } else {
                std::get<tiered_storage>(storage_).for_each_match(
                    [&](const record_type& record) {
                        return equal_(record_value(record), value);
                    },
                    [&](size_type index, const record_type&) {
                        result.push_back(index);
                    });
            }
        }
        return result;
    }

    // Fast unordered duplicate lookup.  Returning IDs avoids the O(k log k)
    // logical-order reconstruction and sort required by find_all().
    [[nodiscard]] std::vector<stable_id> find_all_ids(const T& value) const
        requires(HashIndexEnabled) {
        std::vector<stable_id> result;
        const auto* entry = index_state_.values.find(value);
        if (!entry) {
            return result;
        }
        result.reserve(entry->count);
        auto id = entry->head;
        while (id != invalid_internal_id) {
            result.push_back(stable_id_for_slot(id));
            id = index_state_.ids[static_cast<size_type>(id)].next;
        }
        return result;
    }

    bool erase_one(const T& value) {
        const auto found = find_one(value);
        if (!found) {
            return false;
        }
        erase(*found);
        return true;
    }

    size_type erase_all(const T& value) {
        if constexpr (HashIndexEnabled) {
            const auto* entry = index_state_.values.find(value);
            if (!entry) {
                return 0;
            }
            std::vector<stable_id> ids;
            ids.reserve(entry->count);
            auto id = entry->head;
            while (id != invalid_internal_id) {
                ids.push_back(stable_id_for_slot(id));
                id = index_state_.ids[static_cast<size_type>(id)].next;
            }
            for (const auto current : ids) {
                erase_by_id(current);
            }
            return ids.size();
        } else {
            size_type removed = 0;
            for (size_type i = 0; i < size();) {
                if (equal_(value_at_unchecked(i), value)) {
                    erase(i);
                    ++removed;
                } else {
                    ++i;
                }
            }
            return removed;
        }
    }

    [[nodiscard]] stable_id id_at(size_type index) const
        requires(HashIndexEnabled) {
        check_index(index);
        return stable_id_for_slot(record_at_unchecked(index).id);
    }

    [[nodiscard]] bool id_alive(stable_id id) const noexcept
        requires(HashIndexEnabled) {
        if (id == invalid_id) {
            return false;
        }
        const auto slot = slot_from_stable_id(id);
        if (slot == invalid_internal_id
            || static_cast<size_type>(slot) >= index_state_.ids.size()) {
            return false;
        }
        const auto& metadata = index_state_.ids[static_cast<size_type>(slot)];
        return metadata.generation == generation_from_stable_id(id)
            && (metadata.flags & alive_flag) != 0;
    }

    void erase_by_id(stable_id id)
        requires(HashIndexEnabled) {
        erase(resolve_id(id));
    }

    void force_vector_mode() {
        const auto previous_residency = residency_;
        residency_ = ResidencyMode::forced_vector;
        try {
            convert_to(StorageMode::vector);
        } catch (...) {
            residency_ = previous_residency;
            throw;
        }
    }

    void force_tiered_mode() {
        const auto previous_residency = residency_;
        residency_ = ResidencyMode::forced_tiered;
        try {
            convert_to(StorageMode::tiered);
        } catch (...) {
            residency_ = previous_residency;
            throw;
        }
    }

    void force_tiered_mode(TieredConfig config) {
        const auto previous_residency = residency_;
        const auto previous_config = tiered_config_;
        residency_ = ResidencyMode::forced_tiered;
        try {
            if (mode() == StorageMode::tiered) {
                reconfigure_tiered(config);
            } else {
                tiered_config_ = config;
                convert_to(StorageMode::tiered);
            }
        } catch (...) {
            residency_ = previous_residency;
            tiered_config_ = previous_config;
            throw;
        }
        // on_transition(from, to, active_config) already synchronizes the
        // built-in policy.  Keep this compatibility hook after the successful
        // storage commit so a failed rebuild cannot make policy and storage
        // disagree, and so same-mode telemetry can compare old and new shapes.
        if constexpr (requires(AdaptationPolicy& p) { p.set_tiered_config(config); }) {
            policy_.set_tiered_config(tiered_config_);
        }
    }

    void enable_auto_mode() noexcept { residency_ = ResidencyMode::automatic; }

    void set_read_adaptation_mode(ReadAdaptationMode mode) noexcept {
        read_adaptation_ = mode;
    }

    [[nodiscard]] ReadAdaptationMode read_adaptation_mode() const noexcept {
        return read_adaptation_;
    }

    // Automatic rebuilds are capacity-boundary operations.  This compatibility
    // hook therefore never changes the representation between two resize
    // events; explicit force_* calls remain available for callers that need an
    // immediate manual conversion.
    bool adapt_now() noexcept { return false; }

    [[nodiscard]] T* data() noexcept
        requires(!HashIndexEnabled) {
        if (mode() != StorageMode::vector) {
            return nullptr;
        }
        return std::get<vector_storage>(storage_).data();
    }

    [[nodiscard]] const T* data() const noexcept
        requires(!HashIndexEnabled) {
        if (mode() != StorageMode::vector) {
            return nullptr;
        }
        return std::get<vector_storage>(storage_).data();
    }

    [[nodiscard]] std::optional<std::span<const T>> try_contiguous_view() const noexcept
        requires(!HashIndexEnabled) {
        if (mode() != StorageMode::vector) {
            return std::nullopt;
        }
        const auto& values = std::get<vector_storage>(storage_);
        return std::span<const T>(values.data(), values.size());
    }

    [[nodiscard]] std::span<const T> make_contiguous()
        requires(!HashIndexEnabled) {
        force_vector_mode();
        const auto& values = std::get<vector_storage>(storage_);
        return {values.data(), values.size()};
    }

    [[nodiscard]] std::vector<T> contiguous_copy() const {
        std::vector<T> result;
        result.reserve(size());
        for_each([&](const T& value) { result.push_back(value); });
        return result;
    }

    template <class Function>
    void for_each(Function&& function) {
        if (mode() == StorageMode::vector) {
            for (auto& record : std::get<vector_storage>(storage_)) {
                function(record_value(record));
            }
        } else {
            std::get<tiered_storage>(storage_).for_each(
                [&](record_type& record) { function(record_value(record)); });
        }
        observe_sequential_read(size());
    }

    template <class Function>
    void for_each(Function&& function) const {
        if (mode() == StorageMode::vector) {
            for (const auto& record : std::get<vector_storage>(storage_)) {
                function(record_value(record));
            }
        } else {
            std::get<tiered_storage>(storage_).for_each(
                [&](const record_type& record) { function(record_value(record)); });
        }
        observe_sequential_read(size());
    }

    iterator begin() noexcept { return iterator(this, 0); }
    iterator end() noexcept { return iterator(this, size()); }
    const_iterator begin() const noexcept { return const_iterator(this, 0); }
    const_iterator end() const noexcept { return const_iterator(this, size()); }
    const_iterator cbegin() const noexcept { return begin(); }
    const_iterator cend() const noexcept { return end(); }

    [[nodiscard]] std::size_t allocated_bytes() const noexcept {
        std::size_t bytes = sizeof(*this);
        if (mode() == StorageMode::vector) {
            bytes += std::get<vector_storage>(storage_).capacity() * sizeof(record_type);
        } else {
            bytes += std::get<tiered_storage>(storage_).allocated_bytes();
        }
        if constexpr (HashIndexEnabled) {
            bytes += index_state_.values.allocated_bytes();
            bytes += index_state_.ids.capacity() * sizeof(IdMetadata);
        }
        return bytes;
    }

    [[nodiscard]] std::size_t hash_bucket_count() const noexcept
        requires(HashIndexEnabled) {
        return index_state_.values.bucket_count();
    }

    [[nodiscard]] AdaptationPolicy& policy() noexcept { return policy_; }
    [[nodiscard]] const AdaptationPolicy& policy() const noexcept { return policy_; }

private:
    void reset_after_move() noexcept {
        storage_.template emplace<vector_storage>();
        if constexpr (HashIndexEnabled) {
            index_state_.values.reset_after_move();
            index_state_.ids.clear();
            index_state_.free_head = invalid_internal_id;
        }
        residency_ = ResidencyMode::automatic;
        logical_capacity_ = 0;
        read_sample_countdown_ = read_sample_rate_;
        edit_sample_countdown_ = edit_sample_rate_;
        ++generation_;
    }

    static void set_location(IdMetadata& metadata,
                             bool tiered,
                             std::uint32_t primary,
                             std::uint32_t secondary) noexcept {
        metadata.primary = primary;
        metadata.secondary = secondary;
        if (tiered) {
            metadata.flags |= tiered_flag;
        } else {
            metadata.flags &= static_cast<std::uint8_t>(~tiered_flag);
        }
    }

    static T& record_value(record_type& record) noexcept {
        if constexpr (HashIndexEnabled) {
            return record.value;
        } else {
            return record;
        }
    }

    static const T& record_value(const record_type& record) noexcept {
        if constexpr (HashIndexEnabled) {
            return record.value;
        } else {
            return record;
        }
    }

    static internal_id record_id(const record_type& record) noexcept {
        if constexpr (HashIndexEnabled) {
            return record.id;
        } else {
            (void)record;
            return invalid_internal_id;
        }
    }

    static constexpr unsigned stable_id_slot_bits =
        std::numeric_limits<internal_id>::digits;
    static_assert(stable_id_slot_bits == 32);

    [[nodiscard]] static internal_id slot_from_stable_id(stable_id id) noexcept {
        return static_cast<internal_id>(id);
    }

    [[nodiscard]] static std::uint32_t generation_from_stable_id(
        stable_id id) noexcept {
        return static_cast<std::uint32_t>(id >> stable_id_slot_bits);
    }

    [[nodiscard]] stable_id stable_id_for_slot(internal_id slot) const noexcept
        requires(HashIndexEnabled) {
        const auto generation =
            index_state_.ids[static_cast<size_type>(slot)].generation;
        return (static_cast<stable_id>(generation) << stable_id_slot_bits)
             | static_cast<stable_id>(slot);
    }

    [[nodiscard]] internal_id acquire_id_slot()
        requires(HashIndexEnabled) {
        if (index_state_.free_head != invalid_internal_id) {
            const auto id = index_state_.free_head;
            auto& metadata = index_state_.ids[static_cast<size_type>(id)];
            index_state_.free_head = metadata.free_next;
            assert(metadata.generation
                   < std::numeric_limits<std::uint32_t>::max());
            const auto next_generation = metadata.generation + 1;
            metadata = {};
            metadata.generation = next_generation;
            return id;
        }

        if (index_state_.ids.size() >= static_cast<size_type>(invalid_internal_id)) {
            throw std::length_error("stable id slot space exhausted");
        }
        const auto id = static_cast<internal_id>(index_state_.ids.size());
        index_state_.ids.emplace_back();
        return id;
    }

    void release_id_slot(internal_id id) noexcept
        requires(HashIndexEnabled) {
        assert(id != invalid_internal_id);
        assert(static_cast<size_type>(id) < index_state_.ids.size());
        auto& metadata = index_state_.ids[static_cast<size_type>(id)];
        const auto generation = metadata.generation;
        metadata = {};
        metadata.generation = generation;
        if (generation != std::numeric_limits<std::uint32_t>::max()) {
            metadata.free_next = index_state_.free_head;
            index_state_.free_head = id;
        }
    }

    void rebuild_free_id_list() noexcept
        requires(HashIndexEnabled) {
        index_state_.free_head = invalid_internal_id;
        for (size_type index = index_state_.ids.size(); index != 0; --index) {
            const auto id = static_cast<internal_id>(index - 1);
            auto& metadata = index_state_.ids[index - 1];
            const auto generation = metadata.generation;
            metadata = {};
            metadata.generation = generation;
            if (generation != std::numeric_limits<std::uint32_t>::max()) {
                metadata.free_next = index_state_.free_head;
                index_state_.free_head = id;
            }
        }
    }

    record_type make_record(T value) {
        if constexpr (HashIndexEnabled) {
            const auto id = acquire_id_slot();
            try {
                return IndexedRecord{std::move(value), id};
            } catch (...) {
                release_id_slot(id);
                throw;
            }
        } else {
            return value;
        }
    }

    record_type& record_at_unchecked(size_type index) noexcept {
        return mode() == StorageMode::vector
            ? std::get<vector_storage>(storage_)[index]
            : std::get<tiered_storage>(storage_)[index];
    }

    const record_type& record_at_unchecked(size_type index) const noexcept {
        return mode() == StorageMode::vector
            ? std::get<vector_storage>(storage_)[index]
            : std::get<tiered_storage>(storage_)[index];
    }

    T& value_at_unchecked(size_type index) noexcept {
        return record_value(record_at_unchecked(index));
    }

    const T& value_at_unchecked(size_type index) const noexcept {
        return record_value(record_at_unchecked(index));
    }

    void check_index(size_type index) const {
        if (index >= size()) {
            throw std::out_of_range("AdaptiveSequence index out of range");
        }
    }

    void push_back_impl(T value) {
        const auto old_size = size();
        grow_before_insert(old_size + 1);
        auto record = make_record(std::move(value));
        const auto id = record_id(record);
        auto relocate = relocation_callback();
        if constexpr (HashIndexEnabled) {
            try {
                // Keep an unmoved lookup key so a failed storage insertion can
                // roll the duplicate chain back before the slot is recycled.
                T rollback_key(record.value);
                link_value(record.value, id);
                try {
                    if (mode() == StorageMode::vector) {
                        auto& values = std::get<vector_storage>(storage_);
                        values.push_back(std::move(record));
                        refresh_vector_locations(old_size);
                    } else {
                        std::get<tiered_storage>(storage_).push_back(
                            std::move(record), relocate);
                    }
                } catch (...) {
                    unlink_value(rollback_key, id);
                    throw;
                }
                index_state_.ids[static_cast<size_type>(id)].flags |= alive_flag;
            } catch (...) {
                release_id_slot(id);
                throw;
            }
        } else if (mode() == StorageMode::vector) {
            std::get<vector_storage>(storage_).push_back(std::move(record));
        } else {
            std::get<tiered_storage>(storage_).push_back(std::move(record), relocate);
        }
        ++generation_;
        policy_.observe({OperationKind::append, old_size, old_size, 1, sizeof(T)});
    }

    void insert_impl(size_type index, T value) {
        if (index > size()) {
            throw std::out_of_range("AdaptiveSequence insertion index out of range");
        }
        const auto old_size = size();
        grow_before_insert(old_size + 1);
        auto record = make_record(std::move(value));
        const auto id = record_id(record);
        auto relocate = relocation_callback();
        if constexpr (HashIndexEnabled) {
            try {
                T rollback_key(record.value);
                link_value(record.value, id);
                try {
                    if (mode() == StorageMode::vector) {
                        auto& values = std::get<vector_storage>(storage_);
                        values.insert(values.begin() + static_cast<difference_type>(index),
                                      std::move(record));
                        refresh_vector_locations(index);
                    } else {
                        std::get<tiered_storage>(storage_).insert(
                            index, std::move(record), relocate);
                    }
                } catch (...) {
                    unlink_value(rollback_key, id);
                    throw;
                }
                index_state_.ids[static_cast<size_type>(id)].flags |= alive_flag;
            } catch (...) {
                release_id_slot(id);
                throw;
            }
        } else if (mode() == StorageMode::vector) {
            auto& values = std::get<vector_storage>(storage_);
            values.insert(values.begin() + static_cast<difference_type>(index),
                          std::move(record));
        } else {
            std::get<tiered_storage>(storage_).insert(index, std::move(record), relocate);
        }
        ++generation_;
        observe_structural_edit(OperationKind::insert, old_size, index);
    }

    void set_impl(size_type index, T value) {
        check_index(index);
        auto& record = record_at_unchecked(index);
        if constexpr (HashIndexEnabled) {
            if (equal_(record.value, value)) {
                record.value = std::move(value);
            } else {
                const auto id = record.id;
                unlink_value(record.value, id);
                record.value = std::move(value);
                link_value(record.value, id);
            }
        } else {
            record = std::move(value);
        }
        policy_.observe({OperationKind::set, size(), index, 1, sizeof(T)});
    }

    auto relocation_callback() {
        return [this](const record_type& record,
                      typename tiered_storage::leaf_id_type leaf,
                      size_type local) {
            if constexpr (HashIndexEnabled) {
                auto& metadata = index_state_.ids[static_cast<size_type>(record.id)];
                set_location(metadata, true, static_cast<std::uint32_t>(leaf),
                             static_cast<std::uint32_t>(local));
            } else {
                (void)record;
                (void)leaf;
                (void)local;
            }
        };
    }

    void refresh_vector_locations(size_type first) noexcept {
        if constexpr (HashIndexEnabled) {
            const auto& values = std::get<vector_storage>(storage_);
            for (auto i = first; i < values.size(); ++i) {
                set_location(index_state_.ids[static_cast<size_type>(values[i].id)],
                             false, static_cast<std::uint32_t>(i), 0);
            }
        } else {
            (void)first;
        }
    }

    void link_value(const T& value, internal_id id)
        requires(HashIndexEnabled) {
        auto& entry = index_state_.values.ensure(value);
        auto& metadata = index_state_.ids[static_cast<size_type>(id)];
        metadata.previous = invalid_internal_id;
        metadata.next = entry.head;
        if (entry.head != invalid_internal_id) {
            index_state_.ids[static_cast<size_type>(entry.head)].previous = id;
        }
        entry.head = id;
        ++entry.count;
    }

    void unlink_value(const T& value, internal_id id)
        requires(HashIndexEnabled) {
        auto* entry = index_state_.values.find(value);
        if (!entry) {
            throw std::logic_error("hash index invariant violated");
        }
        auto& metadata = index_state_.ids[static_cast<size_type>(id)];
        if (metadata.previous != invalid_internal_id) {
            index_state_.ids[static_cast<size_type>(metadata.previous)].next = metadata.next;
        } else {
            entry->head = metadata.next;
        }
        if (metadata.next != invalid_internal_id) {
            index_state_.ids[static_cast<size_type>(metadata.next)].previous = metadata.previous;
        }
        metadata.previous = invalid_internal_id;
        metadata.next = invalid_internal_id;
        --entry->count;
        if (entry->count == 0) {
            index_state_.values.erase_key(value);
        }
    }

    [[nodiscard]] size_type resolve_id(stable_id id) const
        requires(HashIndexEnabled) {
        if (id == invalid_id) {
            throw std::out_of_range("unknown stable id");
        }
        const auto slot = slot_from_stable_id(id);
        if (slot == invalid_internal_id
            || static_cast<size_type>(slot) >= index_state_.ids.size()) {
            throw std::out_of_range("unknown stable id");
        }
        const auto& metadata = index_state_.ids[static_cast<size_type>(slot)];
        if (metadata.generation != generation_from_stable_id(id)
            || (metadata.flags & alive_flag) == 0) {
            throw std::out_of_range("stable id no longer refers to an element");
        }
        return resolve_internal_id(slot);
    }

    [[nodiscard]] size_type resolve_internal_id(internal_id id) const noexcept
        requires(HashIndexEnabled) {
        assert(id != invalid_internal_id);
        assert(static_cast<size_type>(id) < index_state_.ids.size());
        const auto& metadata = index_state_.ids[static_cast<size_type>(id)];
        assert((metadata.flags & alive_flag) != 0);
        if ((metadata.flags & tiered_flag) == 0) {
            return metadata.primary;
        }
        return std::get<tiered_storage>(storage_).logical_index(
            static_cast<typename tiered_storage::leaf_id_type>(metadata.primary),
            metadata.secondary);
    }

    [[nodiscard]] static size_type ceil_sqrt(size_type value) noexcept {
        if (value <= 1) {
            return value;
        }
        auto root = static_cast<size_type>(std::sqrt(static_cast<long double>(value)));
        while (root > value / root) {
            --root;
        }
        while (root < value / root
               || (root <= value / root && root * root < value)) {
            ++root;
        }
        return root;
    }

    [[nodiscard]] size_type minimum_tiered_size() const noexcept {
        if constexpr (requires(const AdaptationPolicy& p) {
                          p.config().minimum_tiered_size;
                      }) {
            return std::max<size_type>(1, policy_.config().minimum_tiered_size);
        } else {
            return 4 * 1024;
        }
    }

    [[nodiscard]] size_type shrink_denominator() const noexcept {
        if constexpr (requires(const AdaptationPolicy& p) {
                          p.config().shrink_denominator;
                      }) {
            return std::max<size_type>(2, policy_.config().shrink_denominator);
        } else {
            return 8;
        }
    }

    [[nodiscard]] StorageMode automatic_mode_for(size_type element_count) const noexcept {
        return element_count >= minimum_tiered_size()
            ? StorageMode::tiered : StorageMode::vector;
    }

    [[nodiscard]] TieredConfig geometry_for(size_type element_count) const noexcept {
        auto result = tiered_config_;
        result.leaf_capacity = std::max<size_type>(4, ceil_sqrt(element_count));
        return result;
    }

    void grow_before_insert(size_type required) {
        if (required <= logical_capacity_) {
            return;
        }
        size_type next = logical_capacity_ == 0 ? 1 : logical_capacity_;
        while (next < required) {
            if (next > std::numeric_limits<size_type>::max() / 2) {
                next = required;
                break;
            }
            next *= 2;
        }
        rebuild_for_capacity(next, required);
    }

    void shrink_after_erase() noexcept {
        if (logical_capacity_ <= 1
            || size() > logical_capacity_ / shrink_denominator()) {
            return;
        }
        const auto next = std::max(size(), logical_capacity_ / 2);
        // The erase has already committed.  Capacity shrink is an automatic
        // optimization and must not turn that successful erase into a thrown
        // operation.  Copyable records keep their source intact on failure;
        // nothrow-movable records use preallocated rebuild paths below.
        if constexpr (std::is_copy_constructible_v<record_type>
                      || std::is_nothrow_move_constructible_v<record_type>) {
            try {
                rebuild_for_capacity(next, size());
            } catch (...) {
                // Keep the current (or already committed) capacity/layout and
                // retry at a later resize boundary.
            }
        }
    }

    void rebuild_for_capacity(size_type target_capacity, size_type target_size) {
        target_capacity = std::max(target_capacity, target_size);
        const bool growing = target_capacity >= logical_capacity_;
        if constexpr (HashIndexEnabled) {
            if (growing) {
                index_state_.values.reserve_for_elements(target_capacity);
                index_state_.ids.reserve(
                    std::max(index_state_.ids.size(), target_capacity));
            }
        }
        auto target_mode = mode();
        if (residency_ == ResidencyMode::forced_vector) {
            target_mode = StorageMode::vector;
        } else if (residency_ == ResidencyMode::forced_tiered) {
            target_mode = StorageMode::tiered;
        } else {
            target_mode = automatic_mode_for(target_size);
        }

        if (target_mode == StorageMode::vector) {
            vector_storage rebuilt;
            rebuilt.reserve(target_capacity);
            if (mode() == StorageMode::vector) {
                auto& source = std::get<vector_storage>(storage_);
                if constexpr (std::is_copy_constructible_v<record_type>) {
                    rebuilt.insert(rebuilt.end(), source.begin(), source.end());
                } else {
                    rebuilt.insert(rebuilt.end(),
                        std::make_move_iterator(source.begin()),
                        std::make_move_iterator(source.end()));
                }
            } else if constexpr (std::is_copy_constructible_v<record_type>) {
                const auto& source = std::get<tiered_storage>(storage_);
                source.for_each([&](const record_type& record) {
                    rebuilt.push_back(record);
                });
            } else {
                auto& source = std::get<tiered_storage>(storage_);
                source.for_each([&](record_type& record) {
                    rebuilt.push_back(std::move(record));
                });
            }

            const auto source_mode = mode();
            storage_.template emplace<vector_storage>(std::move(rebuilt));
            logical_capacity_ = target_capacity;
            refresh_vector_locations(0);
            ++generation_;
            if (source_mode != StorageMode::vector) {
                notify_policy_transition(source_mode, StorageMode::vector);
            }
        } else {
            const auto target_config = geometry_for(target_size);
            tiered_storage rebuilt;
            if (mode() == StorageMode::vector) {
                if constexpr (std::is_copy_constructible_v<record_type>) {
                    rebuilt = tiered_storage::from_vector_copy(
                        std::get<vector_storage>(storage_), target_config);
                } else {
                    rebuilt = tiered_storage::from_vector(
                        std::move(std::get<vector_storage>(storage_)), target_config);
                }
            } else if constexpr (std::is_copy_constructible_v<record_type>) {
                rebuilt = tiered_storage::reconfigured_copy(
                    std::get<tiered_storage>(storage_), target_config);
            } else {
                rebuilt = tiered_storage::reconfigured_move(
                    std::get<tiered_storage>(storage_), target_config);
            }

            const auto source_mode = mode();
            tiered_config_ = rebuilt.config();
            storage_.template emplace<tiered_storage>(std::move(rebuilt));
            logical_capacity_ = target_capacity;
            refresh_tiered_locations();
            ++generation_;
            notify_policy_transition(source_mode, StorageMode::tiered);
        }

        if constexpr (HashIndexEnabled) {
            if (!growing) {
                index_state_.values.reserve_for_elements(target_capacity);
            }
        }
    }

    void convert_to(StorageMode target) {
        const auto source = mode();
        if (source == target) {
            return;
        }
        if (target == StorageMode::tiered) {
            tiered_storage tiered;
            if constexpr (std::is_copy_constructible_v<record_type>) {
                tiered = tiered_storage::from_vector_copy(
                    std::get<vector_storage>(storage_), tiered_config_);
            } else {
                tiered = tiered_storage::from_vector(
                    std::move(std::get<vector_storage>(storage_)), tiered_config_);
            }
            tiered_config_ = tiered.config();
            storage_.template emplace<tiered_storage>(std::move(tiered));
            refresh_tiered_locations();
        } else {
            vector_storage values;
            if constexpr (std::is_copy_constructible_v<record_type>) {
                values = std::get<tiered_storage>(storage_).to_vector_copy();
            } else {
                values = std::get<tiered_storage>(storage_).to_vector_move();
            }
            if (logical_capacity_ > values.capacity()) {
                values.reserve(logical_capacity_);
            }
            storage_.template emplace<vector_storage>(std::move(values));
            refresh_vector_locations(0);
        }
        ++generation_;
        notify_policy_transition(source, target);
    }

    void reconfigure_tiered(TieredConfig target_config) {
        if (mode() != StorageMode::tiered) {
            tiered_config_ = target_config;
            return;
        }
        if (target_config == tiered_config_) {
            return;
        }

        if (target_config.leaf_capacity == tiered_config_.leaf_capacity) {
            auto& tiered = std::get<tiered_storage>(storage_);
            tiered.reconfigure_directory(target_config.directory_fanout,
                                         target_config.directory_levels);
            tiered_config_ = tiered.config();
            ++generation_;
            notify_policy_transition(StorageMode::tiered, StorageMode::tiered);
            return;
        }

        tiered_storage rebuilt;
        if constexpr (std::is_copy_constructible_v<record_type>) {
            rebuilt = tiered_storage::reconfigured_copy(
                std::get<tiered_storage>(storage_), target_config);
        } else {
            rebuilt = tiered_storage::reconfigured_move(
                std::get<tiered_storage>(storage_), target_config);
        }
        tiered_config_ = rebuilt.config();
        storage_.template emplace<tiered_storage>(std::move(rebuilt));
        refresh_tiered_locations();
        ++generation_;
        notify_policy_transition(StorageMode::tiered, StorageMode::tiered);
    }

    void notify_policy_transition(StorageMode source, StorageMode target) noexcept {
        if constexpr (requires(AdaptationPolicy& p, StorageMode from,
                               StorageMode to, TieredConfig config) {
                          p.on_transition(from, to, config);
                      }) {
            policy_.on_transition(source, target, tiered_config_);
        } else {
            policy_.on_transition(source, target);
        }
    }

    void refresh_tiered_locations() noexcept {
        if constexpr (HashIndexEnabled) {
            auto relocate = relocation_callback();
            std::get<tiered_storage>(storage_).for_each_with_location(relocate);
        }
    }

    [[nodiscard]] std::size_t configured_read_sample_rate() const noexcept {
        if constexpr (requires(const AdaptationPolicy& p) { p.config().read_sample_rate; }) {
            return std::max<std::size_t>(1, policy_.config().read_sample_rate);
        } else {
            return 256;
        }
    }

    [[nodiscard]] std::size_t configured_edit_sample_rate() const noexcept {
        if constexpr (requires(const AdaptationPolicy& p) { p.config().edit_sample_rate; }) {
            return std::max<std::size_t>(1, policy_.config().edit_sample_rate);
        } else {
            return 1;
        }
    }

    void observe_structural_edit(OperationKind kind,
                                 size_type old_size,
                                 size_type position) noexcept {
        if (--edit_sample_countdown_ == 0) {
            edit_sample_countdown_ = edit_sample_rate_;
            policy_.observe({kind, old_size, position, edit_sample_rate_, sizeof(T)});
        }
    }

    void observe_random_read() const noexcept {
        if (--read_sample_countdown_ == 0) {
            read_sample_countdown_ = read_sample_rate_;
            policy_.observe({OperationKind::random_read, size(), 0, read_sample_rate_, sizeof(T)});
        }
    }

    void observe_sequential_read(size_type count) const noexcept {
        if (count != 0) {
            policy_.observe({OperationKind::sequential_read, size(), 0, count, sizeof(T)});
        }
    }

    TieredConfig tiered_config_;
    mutable AdaptationPolicy policy_;
    std::variant<vector_storage, tiered_storage> storage_;
    [[no_unique_address]] index_state_type index_state_{};
    [[no_unique_address]] Equal equal_{};
    ResidencyMode residency_ = ResidencyMode::automatic;
    ReadAdaptationMode read_adaptation_ = ReadAdaptationMode::deferred;
    size_type logical_capacity_ = 0;
    std::size_t read_sample_rate_ = 256;
    mutable std::size_t read_sample_countdown_ = 256;
    std::size_t edit_sample_rate_ = 1;
    std::size_t edit_sample_countdown_ = 1;
    std::uint64_t generation_ = 0;
};

} // namespace uc