sorted_list.hpp

Go to the documentation of this file.

#pragma once
 
// SortedList<T> — a live-updating sorted view over an ObservableList<T>.
//
// Given a source list and a comparator, SortedList exposes the items of
// the source, reordered to satisfy `comparator(a, b)` as a strict weak
// ordering. Equivalent items (neither `cmp(a,b)` nor `cmp(b,a)` is
// true) retain their source order — SortedList is a STABLE sort, and
// stable at runtime too: no insert, replace, or ItemChanged ever
// permutes two already-equivalent items against each other.
//
// Like FilteredList, every change to the source is translated into the
// minimal incremental sequence of derived events; small source
// mutations never escalate to a full Reset.
//
// Usage:
//
//     auto source = std::make_shared<ObservableList<Task>>();
//     auto sorted = std::make_shared<SortedList<Task>>(
//         source,
//         [](const Task& a, const Task& b) {
//             return a.priority() < b.priority();
//         });
//
//     source->push_back(std::make_shared<Task>(/*prio=*/5));
//     source->push_back(std::make_shared<Task>(/*prio=*/2));
//     // sorted sees: Insert(0), Insert(0)   — the prio=2 ends up at index 0.
//
//     source->at(0)->set_priority(10);
//     // if Task::on_changed propagates, sorted sees Move then ItemChanged
//     // so downstream views refresh the same object at its new position.
//
// Event-translation contract (mandatory; pinned by tests):
//
//     Source event            Derived behaviour
//     ────────────────        ──────────────────────────────────────────
//     Insert(i, x)            binary_search → Insert(j, x)  (always 1 event)
//     Remove(i)               Remove(j)                     (always 1 event)
//     Replace(i, x_new)       d_new == d_old → Replace(d_old, x_new)
//                             otherwise      → Remove(d_old), Insert(d_new)
//     ItemChanged(i)          d_new == d_old → ItemChanged(d_old)
//                             otherwise      → Move(d_old, d_new, x),
//                                              ItemChanged(d_new)
//     Move(from, to)          no event for distinct keys; equivalent keys
//                             follow the new source order using Moves.
//     Reset                   Reset
//
// Objects may mutate before their notifications arrive, including several
// distinct objects in one graph batch. Binary-search inputs are validated in
// O(n); an unordered remainder uses a complete stable permutation repair.
// Repair emits owning Moves (possibly for unchanged rows) plus the primary
// content/structural event in replay order. It costs O(n log n + n*m), where
// m is the number of Moves; ordinary ordered edits remain incremental.
//
// set_comparator() reshuffles the derived list against the new
// comparator and emits a single Reset event (rather than trying to
// emit a minimal-edit sequence — that would be O(n log n) to compute
// and almost always dominated by the re-sort itself). The `reset`
// event is the honest signal "layout changed wholesale".
//
// Stability contract:
//   Equivalent items (by the comparator) are ordered by ascending
//   source index, including after a source Move changes those indices.
//
// Thread-safety:
//   Same policy as FilteredList: internal `shared_mutex`, source
//   listener holds only weak_ptrs, emits happen after the mutation is
//   visible but before the lock is re-taken by a later event.
 
#include "aria/inplace_function.hpp"
#include "aria/list_source.hpp"
#include "aria/observable_list.hpp"
#include "aria/subscription.hpp"
#include "aria/detail/list_signal_mixin.hpp"
#include "aria/detail/typed_signal.hpp"
#include "aria/detail/list_replay.hpp"
#include "aria/detail/list_update_queue.hpp"
 
#include <algorithm>
#include <cstddef>
#include <functional>
#include <type_traits>
#include <memory>
#include <mutex>
#include <optional>
#include <shared_mutex>
#include <stdexcept>
#include <utility>
#include <vector>
 
namespace aria {
 
template<typename T, typename Source = ObservableList<T>>
    requires ListSourceOf<Source, T>
class SortedList
    : public detail::ListSignalMixin<SortedList<T, Source>, T> {
    friend detail::ListSignalMixin<SortedList<T, Source>, T>;
 
public:
    using value_type = T;
    using Comparator = aria::inplace_function<bool(const T&, const T&), 32>;
    using Signal     = detail::ListSignal<T>;
 
private:
    // Erase the whole order scan rather than every adjacent comparison.
    // The scalar adapter and scan refer to the same owned callable instance.
    struct ComparatorTag {};
    struct ComparatorState {
        using Scan = bool (*)(ComparatorState&, const std::vector<std::shared_ptr<T>>&,
                              const std::vector<std::size_t>&, std::optional<std::size_t>);
        Comparator scalar;
        void* target = nullptr;
        Scan scan = nullptr;
        bool operator()(const T& a, const T& b) const { return scalar(a, b); }
    };
 
    template<typename Fn>
    static bool scan_order(Fn& comparator, const std::vector<std::shared_ptr<T>>& items,
                            const std::vector<std::size_t>& sources,
                            std::optional<std::size_t> skip) {
        const auto size = items.size();
        const auto skipped = skip.value_or(size);
        const std::size_t first = skipped == 0 ? 1 : 0;
        if (first >= size) { return true; }
        const T* previous = items[first].get();
        auto previous_source = sources[first];
        for (auto d = first + 1; d < size; ++d) {
            if (d == skipped) { continue; }
            const T* current = items[d].get();
            const auto current_source = sources[d];
            const bool reversed = current_source < previous_source;
            const T* lhs = reversed ? previous : current;
            const T* rhs = reversed ? current : previous;
            if (static_cast<bool>(std::invoke(comparator, *lhs, *rhs)) != reversed) { return false; }
            previous = current;
            previous_source = current_source;
        }
        return true;
    }
 
    static std::shared_ptr<ComparatorState> make_comparator(Comparator comparator) {
        auto result = std::make_shared<ComparatorState>();
        result->scalar = std::move(comparator);
        result->scan = [](ComparatorState& self, const auto& items, const auto& sources, auto skip) {
            return scan_order(self.scalar, items, sources, skip);
        };
        return result;
    }
 
    template<typename Fn>
    static std::shared_ptr<ComparatorState> make_comparator(Fn&& comparator) {
        using Stored = std::decay_t<Fn>;
        // Retain the public small-callable size/alignment contract.
        static_assert(sizeof(Stored) <= 32);
        static_assert(alignof(Stored) <= alignof(std::max_align_t));
        struct OwnedComparator {
            ComparatorState state;
            Stored callable;
            // Fn is the enclosing forwarding-reference type, possibly an lvalue.
            // NOLINTNEXTLINE(cppcoreguidelines-rvalue-reference-param-not-moved)
            explicit OwnedComparator(Fn&& fn) : callable(std::forward<Fn>(fn)) {}
        };
        auto owned = std::make_shared<OwnedComparator>(std::forward<Fn>(comparator));
        auto& result = owned->state;
        result.target = &owned->callable;
        result.scalar = [target = &owned->callable](const T& a, const T& b) {
            return static_cast<bool>(std::invoke(*target, a, b));
        };
        result.scan = [](ComparatorState& self, const auto& items, const auto& sources, auto skip) {
            return scan_order(*static_cast<Stored*>(self.target), items, sources, skip);
        };
        return std::shared_ptr<ComparatorState>{owned, &owned->state};
    }
 
public:
    SortedList(std::shared_ptr<Source> source, Comparator comparator)
        : SortedList(std::move(source), make_comparator(std::move(comparator)), ComparatorTag{}) {}
 
    template<typename Fn>
        requires (!std::is_same_v<std::decay_t<Fn>, Comparator> &&
                  !std::is_pointer_v<std::decay_t<Fn>> &&
                  std::is_copy_constructible_v<std::decay_t<Fn>> &&
                  std::is_invocable_r_v<bool, std::decay_t<Fn>&, const T&, const T&>)
    SortedList(std::shared_ptr<Source> source, Fn&& comparator)
        : SortedList(std::move(source), make_comparator(std::forward<Fn>(comparator)), ComparatorTag{}) {}
 
private:
    SortedList(std::shared_ptr<Source> source, std::shared_ptr<ComparatorState> comparator, ComparatorTag /*tag*/)
        : source_(std::move(source)),
          signal_(std::make_shared<Signal>()),
          state_(std::make_shared<SharedState>())
    {
        state_->comparator = std::move(comparator);
 
        // Build the initial mapping from the source snapshot. See the
        // FilteredList constructor for why we snapshot before
        // subscribing; concurrent source mutations that arrive
        // mid-construction are then delivered as normal listener
        // callbacks on top of the known-good baseline.
        {
            std::unique_lock lk(state_->m);
            state_->source_items = source_->snapshot();
            rebuild_from_snapshot_unlocked_(*state_, state_->source_items);
        }
 
        // Subscribe on the source. The lambda holds only weak_ptrs.
        std::weak_ptr<SharedState>       weak_state  = state_;
        std::weak_ptr<Signal>            weak_signal = signal_;
        std::weak_ptr<Source> weak_source{source_};
        source_sub_ = source_->observe(
            [weak_state, weak_signal, weak_source](const ListChange<T>& ch) {
                auto st  = weak_state.lock();
                auto sig = weak_signal.lock();
                auto src = weak_source.lock();
                if (!st || !sig || !src) return;
                st->updates.submit([st, sig, ch] { dispatch_source_change_(*st, *sig, ch); });
            });
    }
 
public:
    ~SortedList() = default;
 
    SortedList(const SortedList&)            = delete;
    SortedList& operator=(const SortedList&) = delete;
 
    // ── Read surface ──────────────────────────────────────────────────
    [[nodiscard]] std::size_t size() const {
        std::shared_lock lk(state_->m);
        return state_->items.size();
    }
 
    [[nodiscard]] bool empty() const { return size() == 0; }
 
    [[nodiscard]] std::shared_ptr<T> at(std::size_t derived_index) const {
        std::shared_lock lk(state_->m);
        return state_->items.at(derived_index);
    }
 
    [[nodiscard]] std::vector<std::shared_ptr<T>> snapshot() const {
        std::shared_lock lk(state_->m);
        return state_->items;
    }
 
    [[nodiscard]] std::optional<std::size_t>
    source_index_of(std::size_t derived_index) const {
        std::shared_lock lk(state_->m);
        if (derived_index >= state_->derived_to_source.size()) return std::nullopt;
        return state_->derived_to_source[derived_index];
    }
 
    // ── Comparator replacement ────────────────────────────────────────
    //
    // Re-sorts the derived layout against the new comparator and emits
    // a single Reset. An O(n log n) minimal-edit sequence could be
    // computed, but the cost would dominate the re-sort itself and the
    // wider Qt/AppKit adapter ecosystem handles Reset cleanly anyway.
    void set_comparator(Comparator new_comparator) {
        replace_comparator(make_comparator(std::move(new_comparator)));
    }
 
    template<typename Fn>
        requires (!std::is_same_v<std::decay_t<Fn>, Comparator> &&
                  !std::is_pointer_v<std::decay_t<Fn>> &&
                  std::is_copy_constructible_v<std::decay_t<Fn>> &&
                  std::is_invocable_r_v<bool, std::decay_t<Fn>&, const T&, const T&>)
    void set_comparator(Fn&& comparator) {
        replace_comparator(make_comparator(std::forward<Fn>(comparator)));
    }
 
private:
    void replace_comparator(std::shared_ptr<ComparatorState> comparator) {
        auto state = state_;
        auto signal = signal_;
        state->updates.submit([state, signal, comparator = std::move(comparator)] {
            SharedState work;
            work.comparator = comparator;
            rebuild_from_snapshot_unlocked_(work, state->source_items);
            auto reset = ListChange<T>::reset(work.items);
            commit(*state, work);
            state->scratch.reset();
            signal->emit(std::move(reset));
        });
    }
 
    // Per-instance state shared with the source-listener lambda.
    struct SharedState {
        mutable std::shared_mutex       m;
        std::shared_ptr<ComparatorState> comparator;
        detail::ListUpdateQueue updates;
        std::vector<std::shared_ptr<T>> source_items;
        bool needs_rebuild = false;
        std::unique_ptr<SharedState> scratch;
        // length == source.size(); value = derived index of that source slot.
        std::vector<std::size_t>        source_to_derived;
        // length == derived.size(); value = source index.
        std::vector<std::size_t>        derived_to_source;
        // parallel to derived_to_source; strong refs for at() / snapshot().
        std::vector<std::shared_ptr<T>> items;
    };
 
    std::shared_ptr<Source> source_;
    std::shared_ptr<Signal>            signal_;
    std::shared_ptr<SharedState>       state_;
    Subscription                       source_sub_;
 
    // Rebuild all three parallel vectors from a fresh source snapshot.
    // Caller must hold the unique lock on `st.m`. `snap` is taken
    // from the source, whose lock is independent of ours.
    static void rebuild_from_snapshot_unlocked_(
        SharedState& st,
        std::vector<std::shared_ptr<T>> snap)
    {
        std::vector<std::size_t> idx(snap.size());
        for (std::size_t i = 0; i < snap.size(); ++i) idx[i] = i;
        std::stable_sort(idx.begin(), idx.end(),
            [&](std::size_t a, std::size_t b) {
                return (*st.comparator)(*snap[a], *snap[b]);
            });
 
        st.source_to_derived.assign(snap.size(), 0);
        st.derived_to_source.assign(idx.size(), 0);
        st.items.clear();
        st.items.reserve(idx.size());
        for (std::size_t d = 0; d < idx.size(); ++d) {
            const std::size_t s = idx[d];
            st.derived_to_source[d] = s;
            st.source_to_derived[s] = d;
            st.items.push_back(snap[s]);
        }
        renumber_s2d_(st);
    }
 
    struct EmissionBuffer {
        std::vector<ListChange<T>> events;
        void emit(ListChange<T> event) { events.push_back(std::move(event)); }
        void emit_batch(std::vector<ListChange<T>> batch) {
            events.reserve(events.size() + batch.size());
            for (auto& event : batch) { events.push_back(std::move(event)); }
        }
    };
 
    static void commit(SharedState& st, SharedState& work) {
        std::unique_lock lock(st.m);
        st.comparator.swap(work.comparator);
        st.source_to_derived.swap(work.source_to_derived);
        st.derived_to_source.swap(work.derived_to_source);
        st.items.swap(work.items);
        st.needs_rebuild = false;
    }
 
    // Comparators run against a private working layout. Readers and callback
    // reentry always see the previous committed layout; all events publish
    // only after the complete update succeeds. The normal sorted update
    // already scans O(n) rows, so staging preserves its asymptotic bound.
    template<typename Value>
    static void reserve_append(std::vector<Value>& values) {
        if (values.size() != values.capacity()) { return; }
        const auto capacity = values.capacity();
        const auto maximum = values.max_size();
        if (capacity == maximum) { throw std::length_error("SortedList capacity exceeded"); }
        auto next = maximum;
        if (capacity == 0) { next = 1; }
        else if (capacity <= maximum / 2) { next = capacity * 2; }
        values.reserve(next);
    }
 
    static void dispatch_source_change_(SharedState& st,
                                        Signal& sig,
                                        const ListChange<T>& ch) {
        detail::replay_list_change(st.source_items, ch);
        const bool rebuild = st.needs_rebuild;
        st.needs_rebuild = true;
        if (!rebuild && ch.kind == ListChangeKind::Insert && ch.index == st.items.size() &&
            ordered_unlocked_(st)) {
            const auto position = binary_search_insert_(st, *ch.item, ch.index);
            std::vector<ListChange<T>> events;
            events.push_back({ListChangeKind::Insert, position, ch.item, 0});
            {
                std::unique_lock lock(st.m);
                // Complete every potentially throwing allocation before the
                // first logical change. Shared-handle and index moves below
                // are non-throwing; readers see either complete layout.
                reserve_append(st.items);
                reserve_append(st.source_to_derived);
                reserve_append(st.derived_to_source);
                st.items.insert(st.items.begin() + static_cast<std::ptrdiff_t>(position), ch.item);
                st.derived_to_source.insert(
                    st.derived_to_source.begin() + static_cast<std::ptrdiff_t>(position), ch.index);
                st.source_to_derived.push_back(position);
                renumber_s2d_(st, position);
                st.needs_rebuild = false;
            }
            // A later non-append update can recreate its transactional cache.
            // Do not retain a stale copy after a directly committed append.
            st.scratch.reset();
            sig.emit_batch(std::move(events));
            return;
        }
        if (!st.scratch) {
            auto scratch = std::make_unique<SharedState>();
            scratch->comparator = st.comparator;
            scratch->source_to_derived = st.source_to_derived;
            scratch->derived_to_source = st.derived_to_source;
            scratch->items = st.items;
            st.scratch = std::move(scratch);
        }
        auto& work = *st.scratch;
        EmissionBuffer buffered;
        if (rebuild) {
            rebuild_from_snapshot_unlocked_(work, st.source_items);
            buffered.emit(ListChange<T>::reset(work.items));
        } else {
            switch (ch.kind) {
            case ListChangeKind::Insert:      handle_insert_(work, buffered, ch);       break;
            case ListChangeKind::Remove:      handle_remove_(work, buffered, ch);       break;
            case ListChangeKind::Replace:     handle_replace_(work, buffered, ch);      break;
            case ListChangeKind::ItemChanged: handle_item_changed_(work, buffered, ch); break;
            case ListChangeKind::Move:        handle_move_(work, buffered, ch);         break;
            case ListChangeKind::Reset:       handle_reset_(work, buffered, ch);        break;
            }
        }
        commit(st, work);
        // After the swap, work contains the previous public layout. Replay
        // the successful delta into it for the next transaction. This reuses
        // capacity and avoids copying every shared_ptr on every source edit.
        // A cache-allocation failure may discard the workspace; the committed
        // update and its notification remain valid.
        try {
            for (const auto& event : buffered.events) {
                detail::replay_list_change(work.items, event);
            }
            work.source_to_derived = st.source_to_derived;
            work.derived_to_source = st.derived_to_source;
            work.comparator = st.comparator;
        } catch (...) {
            st.scratch.reset();
        }
        sig.emit_batch(std::move(buffered.events));
        // Retired rows and comparator captures run outside the public lock.
    }

    static std::size_t binary_search_insert_(
        const SharedState& st, const T& needle, std::size_t needle_src_i,
        std::optional<std::size_t> skip_derived_idx = std::nullopt)
    {
        const auto& cmp   = *st.comparator;
        const auto& items = st.items;
        const auto& d2s   = st.derived_to_source;
 
        // `real` translates from compressed (with d_old skipped) to
        // real derived-index. Only used to access items[] / d2s[]
        // during the search — the returned value stays in compressed
        // coordinates.
        auto real = [&](std::size_t compressed) -> std::size_t {
            if (skip_derived_idx && compressed >= *skip_derived_idx)
                return compressed + 1;
            return compressed;
        };
        const std::size_t n_eff = items.size()
                                  - (skip_derived_idx ? 1u : 0u);
 
        std::size_t lo = 0;
        std::size_t hi = n_eff;
        while (lo < hi) {
            const std::size_t mid = lo + (hi - lo) / 2;
            const std::size_t r   = real(mid);
            const auto& mid_item  = *items[r];
            if (cmp(needle, mid_item)) {
                hi = mid;
            } else if (cmp(mid_item, needle)) {
                lo = mid + 1;
            } else {
                // Equivalent keys — stability tie-breaker by source
                // index.
                if (needle_src_i < d2s[r]) hi = mid;
                else                        lo = mid + 1;
            }
        }
        return lo;
    }
 
    static void handle_insert_(SharedState& st, EmissionBuffer& sig,
                               const ListChange<T>& ch) {
        std::unique_lock lk(st.m);
        const std::size_t src_idx = ch.index;
 
        // Shift all source indices >= src_idx by +1 (the source just
        // grew one slot before/at them).
        if (src_idx != st.items.size()) {
            for (auto& s : st.derived_to_source) {
                if (s >= src_idx) ++s;
            }
        }
        st.source_to_derived.insert(st.source_to_derived.begin()
                                    + static_cast<std::ptrdiff_t>(src_idx),
                                    0);
 
        // Resolve the current shared_ptr from the source. source has
        // already released its own lock by the time emit() runs.
        auto shared = ch.item;
 
        const bool ordered = ordered_unlocked_(st);
        const std::size_t d_idx = ordered ? binary_search_insert_(st, *shared, src_idx) : st.items.size();
 
        st.derived_to_source.insert(st.derived_to_source.begin()
                                    + static_cast<std::ptrdiff_t>(d_idx),
                                    src_idx);
        st.items.insert(st.items.begin()
                        + static_cast<std::ptrdiff_t>(d_idx),
                        shared);
 
        // Rebuild s2d from d2s (the insert shifted all derived
        // indices >= d_idx by +1).
        renumber_s2d_(st, d_idx);
 
        if (!ordered) {
            auto moves = reorder_unlocked_(st);
            moves.insert(moves.begin(), {ListChangeKind::Insert, d_idx, ch.item, 0});
            lk.unlock();
            sig.emit_batch(std::move(moves));
        } else {
            lk.unlock();
            sig.emit(ListChange<T>{ListChangeKind::Insert, d_idx, ch.item, 0});
        }
    }
 
    static void handle_remove_(SharedState& st, EmissionBuffer& sig,
                               const ListChange<T>& ch) {
        std::unique_lock lk(st.m);
        const std::size_t src_idx = ch.index;
        if (src_idx >= st.source_to_derived.size()) return;
 
        const std::size_t d_idx = st.source_to_derived[src_idx];
        auto removed = st.items[d_idx];
 
        st.items.erase(st.items.begin() + static_cast<std::ptrdiff_t>(d_idx));
        st.derived_to_source.erase(
            st.derived_to_source.begin() + static_cast<std::ptrdiff_t>(d_idx));
        st.source_to_derived.erase(
            st.source_to_derived.begin() + static_cast<std::ptrdiff_t>(src_idx));
 
        // All source indices > src_idx shift down by 1.
        for (auto& s : st.derived_to_source) {
            if (s > src_idx) --s;
        }
        renumber_s2d_(st);
 
        if (!ordered_unlocked_(st)) {
            auto moves = reorder_unlocked_(st);
            moves.insert(moves.begin(), {ListChangeKind::Remove, d_idx, removed, 0});
            lk.unlock();
            sig.emit_batch(std::move(moves));
        } else {
            lk.unlock();
            sig.emit(ListChange<T>{ListChangeKind::Remove, d_idx, removed, 0});
        }
    }
 
    static void handle_replace_(SharedState& st, EmissionBuffer& sig,
                                const ListChange<T>& ch) {
        handle_slot_changed_(st, sig, ch,
                             /*new_ptr_from_src=*/true,
                             /*same_slot_kind=*/ListChangeKind::Replace,
                             /*cross_slot_use_move=*/false);
    }
 
    static void handle_item_changed_(SharedState& st, EmissionBuffer& sig,
                                     const ListChange<T>& ch) {
        handle_slot_changed_(st, sig, ch,
                             /*new_ptr_from_src=*/false,
                             /*same_slot_kind=*/ListChangeKind::ItemChanged,
                             /*cross_slot_use_move=*/true);
    }
 
    // Objects can change before their notifications arrive (graph batches,
    // shared handles, or reentrant observers). Never binary-search until the
    // remaining rows have been checked against their current values.
    static bool ordered_unlocked_(const SharedState& st,
                                  std::optional<std::size_t> skip = std::nullopt) {
        return st.comparator->scan(*st.comparator, st.items, st.derived_to_source, skip);
    }
 
    // Move one row by shifting its interval once. Unlike a general rotate,
    // this does not swap shared handles repeatedly around permutation cycles.
    static void move_row_unlocked_(SharedState& st, std::size_t from, std::size_t to) noexcept {
        auto item = std::move(st.items[from]);
        const auto source_index = st.derived_to_source[from];
        const auto first = static_cast<std::ptrdiff_t>(std::min(from, to));
        const auto last = static_cast<std::ptrdiff_t>(std::max(from, to));
        if (from < to) {
            std::move(st.items.begin() + first + 1, st.items.begin() + last + 1, st.items.begin() + first);
            std::move(st.derived_to_source.begin() + first + 1, st.derived_to_source.begin() + last + 1,
                      st.derived_to_source.begin() + first);
        } else {
            std::move_backward(st.items.begin() + first, st.items.begin() + last, st.items.begin() + last + 1);
            std::move_backward(st.derived_to_source.begin() + first, st.derived_to_source.begin() + last,
                               st.derived_to_source.begin() + last + 1);
        }
        st.items[to] = std::move(item);
        st.derived_to_source[to] = source_index;
        for (auto d = std::min(from, to); d <= std::max(from, to); ++d)
            st.source_to_derived[st.derived_to_source[d]] = d;
    }

    static std::vector<ListChange<T>> reorder_unlocked_(SharedState& st) {
        auto before = [&](std::size_t a, std::size_t b) {
            const auto& lhs = *st.items[st.source_to_derived[a]];
            const auto& rhs = *st.items[st.source_to_derived[b]];
            if ((*st.comparator)(lhs, rhs)) return true;
            if ((*st.comparator)(rhs, lhs)) return false;
            return a < b;
        };
        std::vector<ListChange<T>> events;
        if (std::is_sorted(st.derived_to_source.begin(), st.derived_to_source.end(), before)) return events;
        auto target = st.derived_to_source;
        std::sort(target.begin(), target.end(), before);
        events.reserve(target.size());
        // Move the farther-displaced end of the interval into place. This
        // avoids moving a long repeated block one occurrence at a time when
        // shifting the few intervening rows represents the same permutation.
        std::size_t first = 0, last = target.size();
        while (first < last && target[first] == st.derived_to_source[first]) ++first;
        while (last > first && target[last - 1] == st.derived_to_source[last - 1]) --last;
        const bool reverse = st.source_to_derived[target[first]] - first <
                             last - 1 - st.source_to_derived[target[last - 1]];
        auto place = [&](std::size_t to) {
            const auto from = st.source_to_derived[target[to]];
            if (from == to) return;
            const auto item = st.items[from];
            events.push_back({ListChangeKind::Move, to, item, from});
            move_row_unlocked_(st, from, to);
        };
        if (reverse) { for (auto i = last; i-- > first;) place(i); }
        else { for (auto i = first; i < last; ++i) place(i); }
        return events;
    }

    static void handle_slot_changed_(SharedState& st, EmissionBuffer& sig,
                                     const ListChange<T>& ch,
                                     bool new_ptr_from_src,
                                     ListChangeKind same_slot_kind,
                                     bool cross_slot_use_move) {
        std::unique_lock lk(st.m);
        const std::size_t src_idx = ch.index;
        if (src_idx >= st.source_to_derived.size()) return;
 
        const std::size_t d_old = st.source_to_derived[src_idx];
 
        // Replace carries the new owning handle; ItemChanged keeps identity.
        auto previous = st.items[d_old];
        auto fresh = new_ptr_from_src ? ch.item : previous;
        if (new_ptr_from_src) st.items[d_old] = fresh;
        if (!ordered_unlocked_(st, d_old)) {
            auto moves = reorder_unlocked_(st);
            if (new_ptr_from_src) {
                // Replace the old logical row before replaying the permutation.
                moves.insert(moves.begin(), {ListChangeKind::Replace, d_old, fresh, 0});
            } else {
                moves.push_back({ListChangeKind::ItemChanged, st.source_to_derived[src_idx], fresh, 0});
            }
            lk.unlock();
            sig.emit_batch(std::move(moves));
            return;
        }
 
        // Where does the (possibly mutated) item belong now? Skip the
        // old slot in the search so we measure "new position if the
        // old slot weren't there". Returned `p_small` is in the
        // compressed (d_old-skipped) view: `p_small == d_old` means
        // the item belongs right back where it was.
        const std::size_t p_small = binary_search_insert_(
            st, *fresh, src_idx, /*skip_derived_idx=*/d_old);
 
        if (p_small == d_old) {
            // Item stayed in place — one same-slot event.
            lk.unlock();
            sig.emit(ListChange<T>{same_slot_kind, d_old, fresh, 0});
            return;
        }
 
        // Shift only the affected interval. Erase followed by insert would
        // shift the unchanged suffix twice and renumber unaffected rows.
        const std::size_t d_insert = p_small;
        move_row_unlocked_(st, d_old, d_insert);
 
        lk.unlock();
        if (cross_slot_use_move) {
            sig.emit_batch({ListChange<T>{ListChangeKind::Move, d_insert, fresh, d_old},
                            ListChange<T>{ListChangeKind::ItemChanged, d_insert, fresh, 0}});
        } else {
            sig.emit_batch({ListChange<T>{ListChangeKind::Remove, d_old, previous, 0},
                            ListChange<T>{ListChangeKind::Insert, d_insert, fresh, 0}});
        }
    }

    static void handle_move_(SharedState& st, EmissionBuffer& sig,
                             const ListChange<T>& ch) {
        std::unique_lock lk(st.m);
        const std::size_t from = ch.from_index;
        const std::size_t to   = ch.index;
        if (from == to) return;
        if (from >= st.source_to_derived.size()) return;
        if (to   >= st.source_to_derived.size()) return;
 
        // Shift all source indices that lived in [from+1..to] left by 1
        // (if from < to) or all indices in [to..from-1] right by 1
        // (if from > to). The moved item itself gets the new value
        // `to`.
        const std::size_t moved_d = st.source_to_derived[from];
 
        if (from < to) {
            for (auto& s : st.derived_to_source) {
                if (s > from && s <= to) --s;
            }
        } else {
            for (auto& s : st.derived_to_source) {
                if (s >= to && s < from) ++s;
            }
        }
        st.derived_to_source[moved_d] = to;
        renumber_s2d_(st);
        // A source Move normally changes only source-index metadata. Pending
        // object-key writes can also have invalidated the visible key order.
        if (!ordered_unlocked_(st)) {
            auto moves = reorder_unlocked_(st);
            lk.unlock();
            sig.emit_batch(std::move(moves));
        }
    }
 
    static void handle_reset_(SharedState& st, EmissionBuffer& sig,
                              const ListChange<T>& ch) {
        ListChange<T> reset;
        {
            std::unique_lock lk(st.m);
            rebuild_from_snapshot_unlocked_(st, *ch.snapshot);
            reset = ListChange<T>::reset(st.items);
        }
        sig.emit(std::move(reset));
    }

    static void renumber_s2d_(SharedState& st, std::size_t first = 0) {
        for (std::size_t d = first; d < st.derived_to_source.size(); ++d) {
            st.source_to_derived[st.derived_to_source[d]] = d;
        }
    }
};
 
// ---------------------------------------------------------------------------
//  Factory helper — deduces the source type so pipelines stay readable.
//  See the note on `aria::filtered` in filtered_list.hpp.
// ---------------------------------------------------------------------------
template<typename Source,
         typename Comparator,
         typename T = list_source_value_t<Source>>
    requires ListSourceOf<Source, T>
[[nodiscard]] std::shared_ptr<SortedList<T, Source>>
sorted(std::shared_ptr<Source> source, Comparator comparator) {
    return std::make_shared<SortedList<T, Source>>(
        std::move(source),
        std::move(comparator));
}
 
}  // namespace aria