#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 <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 = CostModelPolicy,
          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;
        std::uint32_t primary = 0;   // vector index or stable leaf id
        std::uint32_t secondary = 0; // local leaf index
        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;
    };

    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&&) noexcept = default;
    AdaptiveSequence& operator=(const AdaptiveSequence&) = default;
    AdaptiveSequence& operator=(AdaptiveSequence&&) noexcept = default;
    ~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; }
    [[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) {
        if (read_adaptation_ == ReadAdaptationMode::eager_nonconst) {
            apply_pending_adaptation();
        }
        observe_random_read();
        if (read_adaptation_ == ReadAdaptationMode::eager_nonconst) {
            apply_pending_adaptation();
        }
        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);
        apply_pending_adaptation();
        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) {
            auto& metadata = index_state_.ids[static_cast<size_type>(id)];
            metadata.flags &= static_cast<std::uint8_t>(~alive_flag);
            metadata.previous = invalid_internal_id;
            metadata.next = invalid_internal_id;
        }
        ++generation_;
        observe_structural_edit(OperationKind::erase, old_size, index);
    }

    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) {
            for (auto& metadata : index_state_.ids) {
                metadata.flags &= static_cast<std::uint8_t>(~alive_flag);
                metadata.previous = invalid_internal_id;
                metadata.next = invalid_internal_id;
            }
            index_state_.values.clear();
        }
        storage_.template emplace<vector_storage>();
        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();
        std::get<vector_storage>(storage_).reserve(capacity);
        if constexpr (HashIndexEnabled) {
            const auto additional = capacity > size() ? capacity - 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_id(entry->head);
        } else {
            for (size_type i = 0; i < size(); ++i) {
                if (equal_(value_at_unchecked(i), value)) {
                    return i;
                }
            }
            return std::nullopt;
        }
    }

    [[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_id(id));
                id = index_state_.ids[static_cast<size_type>(id)].next;
            }
            std::sort(result.begin(), result.end());
        } else {
            for (size_type i = 0; i < size(); ++i) {
                if (equal_(value_at_unchecked(i), value)) {
                    result.push_back(i);
                }
            }
        }
        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(static_cast<stable_id>(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<internal_id> ids;
            ids.reserve(entry->count);
            auto id = entry->head;
            while (id != invalid_internal_id) {
                ids.push_back(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 static_cast<stable_id>(record_at_unchecked(index).id);
    }

    [[nodiscard]] bool id_alive(stable_id id) const noexcept
        requires(HashIndexEnabled) {
        return id <= static_cast<stable_id>(std::numeric_limits<internal_id>::max())
            && id < index_state_.ids.size()
            && (index_state_.ids[static_cast<size_type>(id)].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_;
    }

    bool adapt_now() {
        if (residency_ != ResidencyMode::automatic) {
            return false;
        }
        return apply_pending_adaptation();
    }

    [[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]] AdaptationPolicy& policy() noexcept { return policy_; }
    [[nodiscard]] const AdaptationPolicy& policy() const noexcept { return policy_; }

private:
    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;
        }
    }

    record_type make_record(T value) {
        if constexpr (HashIndexEnabled) {
            if (index_state_.ids.size() >= static_cast<size_type>(invalid_internal_id)) {
                throw std::length_error("stable id space exhausted");
            }
            const auto id = static_cast<internal_id>(index_state_.ids.size());
            index_state_.ids.emplace_back();
            return IndexedRecord{std::move(value), id};
        } 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) {
        apply_pending_adaptation();
        const auto old_size = size();
        auto record = make_record(std::move(value));
        const auto id = record_id(record);
        auto relocate = relocation_callback();
        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);
        }
        if constexpr (HashIndexEnabled) {
            auto& metadata = index_state_.ids[static_cast<size_type>(id)];
            metadata.flags |= alive_flag;
            link_value(value_at_unchecked(old_size), id);
        }
        ++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");
        }
        apply_pending_adaptation();
        const auto old_size = size();
        auto record = make_record(std::move(value));
        const auto id = record_id(record);
        auto relocate = relocation_callback();
        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);
        }
        if constexpr (HashIndexEnabled) {
            auto& metadata = index_state_.ids[static_cast<size_type>(id)];
            metadata.flags |= alive_flag;
            link_value(value_at_unchecked(index), id);
        }
        ++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 >= index_state_.ids.size()) {
            throw std::out_of_range("unknown stable id");
        }
        const auto& metadata = index_state_.ids[static_cast<size_type>(id)];
        if ((metadata.flags & alive_flag) == 0) {
            throw std::out_of_range("stable id no longer refers to an element");
        }
        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);
    }

    bool apply_pending_adaptation() {
        if (residency_ != ResidencyMode::automatic || !policy_.decision_ready()) {
            return false;
        }

        if constexpr (requires(AdaptationPolicy& p, StorageMode current,
                               size_type count, TieredConfig config) {
                          p.recommended_decision(current, count, config);
                      }) {
            const auto recommendation =
                policy_.recommended_decision(mode(), size(), tiered_config_);
            if (!recommendation) {
                return false;
            }
            if (recommendation->target == mode()) {
                if (mode() == StorageMode::tiered
                    && recommendation->tiered_config != tiered_config_) {
                    reconfigure_tiered(recommendation->tiered_config);
                    return true;
                }
                return false;
            }
            if (recommendation->target == StorageMode::tiered) {
                const auto previous_config = tiered_config_;
                tiered_config_ = recommendation->tiered_config;
                try {
                    convert_to(recommendation->target);
                } catch (...) {
                    // The recommended shape is policy state, not committed
                    // container state.  A failed allocation/copy must not make
                    // tiered_config() describe storage that was never built.
                    tiered_config_ = previous_config;
                    throw;
                }
                return true;
            }
            convert_to(recommendation->target);
            return true;
        } else {
            const auto recommendation = policy_.recommended_mode(mode(), size());
            if (!recommendation || *recommendation == mode()) {
                return false;
            }
            convert_to(*recommendation);
            return true;
        }
    }

    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 {
                auto values = std::move(std::get<vector_storage>(storage_));
                tiered = tiered_storage::from_vector(std::move(values), 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();
            }
            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 {
            auto values = std::get<tiered_storage>(storage_).to_vector_move();
            rebuilt = tiered_storage::from_vector(std::move(values), 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;
    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