filtered_list.hpp

Go to the documentation of this file.

#pragma once
 
// FilteredList<T> — a live-updating filtered view over an ObservableList<T>.
//
// Given a source list and a predicate, FilteredList exposes the subset of
// items for which the predicate returns true, along with the same
// observation surface (ListChange events, snapshot, at/size). Every change
// to the source is translated into the minimal incremental sequence of
// derived events; small source mutations **never** escalate to a full
// Reset on the derived side.
//
// Usage:
//
//     auto source = std::make_shared<ObservableList<Task>>();
//     auto active = std::make_shared<FilteredList<Task>>(
//         source,
//         [](const Task& t) { return !t.is_done(); });
//
//     auto sub = active->observe([](const ListChange<Task>& ch) {
//         // UI-side handler; receives derived-coordinate events.
//     });
//
//     source->push_back(std::make_shared<Task>("write docs"));
//     // → active sees Insert(0)
//     source->at(0)->mark_done();
//     // → active sees Remove(0)    (if Task exposes on_changed)
//
// Event-translation contract (mandatory; pinned by tests):
//
//     Source event            Derived behaviour
//     ────────────────        ──────────────────────────────────────────
//     Insert(i, x)            p(x) ? Insert(j, x) : nothing
//     Remove(i)               was-in ? Remove(j)  : nothing
//     Replace(i, x_new)       in→in  : Replace(j, x_new)
//                             in→out : Remove(j)
//                             out→in : Insert(j, x_new)
//                             out→out: nothing
//     ItemChanged(i)          in still in  : ItemChanged(j)
//                             in  → out    : Remove(j)
//                             out → in     : Insert(j, <cur>)
//                             out still out: nothing
//     Move(from, to, x)       in→in  : Move(j_from, j_to, x)
//                                      (j_from / j_to recomputed from
//                                      the post-move layout)
//                             items not in the filter produce no event
//     Reset                   always : Reset
//
// set_predicate() applies a membership diff: each source item that
// transitions in↔out produces a single Insert / Remove on the derived
// side. Items that stay in or stay out produce no event.
//
// Thread-safety: the derived list protects its own state with a
// `shared_mutex`. The listener attached to the source is invoked on
// whatever thread the source emits on (ObservableList emits AFTER
// releasing its own mutex, so the listener may call back into the
// source safely). FilteredList itself is graph-thread-agnostic;
// callers who want UI-thread delivery should put a dispatcher between
// the source emission and the FilteredList observer.
//
// Lifetime:
// - FilteredList holds a strong shared_ptr to its source, so the source is
//   guaranteed to outlive the derived view. This holds transitively when
//   sources are chained: each link keeps the one below it alive.
// - The source listener captures a weak_ptr<SharedState> and a
//   weak_ptr<Signal>; destroying the FilteredList mid-emission
//   degrades the listener to a no-op and the Subscription RAII
//   releases the connection cleanly.
//
// Chaining:
// - The source type is a template parameter defaulting to
//   `ObservableList<T>` and constrained to `ListSourceOf<Source, T>`. Any
//   type satisfying that concept works, which is what makes
//   `FilteredList -> SortedList -> PagedList` pipelines possible. Prefer the
//   `filtered()` helper at the bottom of this file so the source type does
//   not have to be spelled out.
 
#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/list_replay.hpp"
#include "aria/detail/list_update_queue.hpp"
#include "aria/detail/typed_signal.hpp"
 
#include <algorithm>
#include <cstddef>
#include <memory>
#include <mutex>
#include <optional>
#include <shared_mutex>
#include <utility>
#include <vector>
 
namespace aria {
 
template<typename T, typename Source = ObservableList<T>>
    requires ListSourceOf<Source, T>
class FilteredList
    : public detail::ListSignalMixin<FilteredList<T, Source>, T> {
    friend detail::ListSignalMixin<FilteredList<T, Source>, T>;
 
public:
    using value_type = T;
    using Predicate  = aria::inplace_function<bool(const T&), 32>;
    using Signal     = detail::ListSignal<T>;
 
    FilteredList(std::shared_ptr<Source> source,
                 Predicate predicate)
        : source_(std::move(source)),
          signal_(std::make_shared<Signal>()),
          state_(std::make_shared<SharedState>())
    {
        state_->predicate = std::make_shared<Predicate>(std::move(predicate));
 
        // Construct on the source writer thread or while its writer is
        // quiescent: snapshot + subscribe is not a cross-thread transaction.
        // The signal watermark excludes an active batch already in snapshot.
        {
            std::unique_lock lk(state_->m);
            auto snap = source_->snapshot();
            state_->source_items = snap;
            state_->source_to_derived.assign(snap.size(), std::nullopt);
            for (std::size_t i = 0; i < snap.size(); ++i) {
                if ((*state_->predicate)(*snap[i])) {
                    state_->source_to_derived[i] = state_->derived_to_source.size();
                    state_->derived_to_source.push_back(i);
                    state_->items.push_back(snap[i]);
                }
            }
        }
 
        // Subscribe on the source. The lambda holds only weak_ptrs so
        // destroying the FilteredList while a notification is in
        // flight is safe — the lock() attempt returns nullptr.
        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); });
            });
    }
 
    ~FilteredList() = default;
 
    FilteredList(const FilteredList&)            = delete;
    FilteredList& operator=(const FilteredList&) = 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];
    }
 
    // ── Predicate replacement (incremental diff) ──────────────────────
    //
    // For each source item, compute the new membership; emit Insert /
    // Remove only for items whose membership changed. Items that stay
    // in or stay out produce no event — if the caller wanted "refresh
    // as if every in-item had changed", they should instead send a
    // proper source-side mutation or iterate ItemChanged manually.
    void set_predicate(Predicate new_predicate) {
        auto state = state_;
        auto signal = signal_;
        auto predicate = std::make_shared<Predicate>(std::move(new_predicate));
        state->updates.submit([state, signal, predicate = std::move(predicate)] {
            replace_predicate(*state, *signal, predicate);
        });
    }
 
private:
    // Per-instance state shared with the source-listener lambda. Held
    // as shared_ptr so the listener never dereferences a dangling
    // control block.
    struct SharedState {
        mutable std::shared_mutex                 m;
        std::shared_ptr<Predicate>                predicate;
        detail::ListUpdateQueue updates;
        bool needs_rebuild = false;
        // length == source.size(); nullopt means "not in derived".
        std::vector<std::optional<std::size_t>>   source_to_derived;
        // length == derived.size(); value is the 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::vector<std::shared_ptr<T>>           source_items;
    };
 
    std::shared_ptr<Source> source_;
    std::shared_ptr<Signal>            signal_;
    std::shared_ptr<SharedState>       state_;
    Subscription                       source_sub_;
 
    template<class Value>
    static void reserve_one(std::vector<Value>& values) {
        if (values.size() != values.capacity()) { return; }
        const auto next = values.size() < values.max_size() / 2
            ? std::max<std::size_t>(1, values.size() * 2) : values.max_size();
        values.reserve(next);
    }
 
    static void replace_predicate(SharedState& st, Signal& signal,
                                    std::shared_ptr<Predicate> predicate) {
        // Only the update queue writes these vectors. Prepare against the
        // current source mirror without holding the public readers' lock.
        const auto& snap = st.source_items;
        std::vector<std::optional<std::size_t>> s2d(snap.size(), std::nullopt);
        std::vector<std::size_t> d2s;
        std::vector<std::shared_ptr<T>> items;
        std::vector<ListChange<T>> events;
        d2s.reserve(snap.size());
        items.reserve(snap.size());
        events.reserve(snap.size());
        std::size_t observed_pos = 0;
        for (std::size_t i = 0; i < snap.size(); ++i) {
            const bool was_in = i < st.source_to_derived.size() && st.source_to_derived[i].has_value();
            const bool is_in = (*predicate)(*snap[i]);
            if (is_in) {
                s2d[i] = items.size();
                d2s.push_back(i);
                items.push_back(snap[i]);
            }
            if (was_in && !is_in) {
                events.push_back({ListChangeKind::Remove, observed_pos, snap[i], 0});
            } else if (!was_in && is_in) {
                events.push_back({ListChangeKind::Insert, observed_pos, snap[i], 0});
            }
            if (is_in) { ++observed_pos; }
        }
        if (st.needs_rebuild) { events = {ListChange<T>::reset(items)}; }
        {
            std::unique_lock lock(st.m);
            st.predicate.swap(predicate);
            st.source_to_derived.swap(s2d);
            st.derived_to_source.swap(d2s);
            st.items.swap(items);
            st.needs_rebuild = false;
        }
        signal.emit_batch(std::move(events));
        // Retired items and callable captures are destroyed without the lock.
    }
 
    // ── Translation: one source event -> zero or one derived events ───
    static void dispatch_source_change_(SharedState& st,
                                        Signal& sig,
                                        const ListChange<T>& ch) {
        detail::replay_list_change(st.source_items, ch);
        if (st.needs_rebuild) {
            handle_reset_(st, sig, ListChange<T>::reset(st.source_items));
            st.needs_rebuild = false;
            return;
        }
        st.needs_rebuild = true;
        switch (ch.kind) {
        case ListChangeKind::Insert:      handle_insert_(st, sig, ch);       break;
        case ListChangeKind::Remove:      handle_remove_(st, sig, ch);       break;
        case ListChangeKind::Replace:     handle_replace_(st, sig, ch);      break;
        case ListChangeKind::ItemChanged: handle_item_changed_(st, sig, ch); break;
        case ListChangeKind::Move:        handle_move_(st, sig, ch);         break;
        case ListChangeKind::Reset:       handle_reset_(st, sig, ch);        break;
        }
        st.needs_rebuild = false;
    }
 
    static void handle_insert_(SharedState& st, Signal& sig,
                               const ListChange<T>& ch) {
        const bool is_in = (*st.predicate)(*ch.item);
        std::unique_lock lk(st.m);
        const std::size_t src_idx = ch.index;
        // Allocate before modifying any parallel index vector.
        reserve_one(st.source_to_derived);
        if (is_in) {
            reserve_one(st.items);
            reserve_one(st.derived_to_source);
        }
        // Source indices are ordered, so this locates both the insertion
        // position and the only suffix whose source indices change. A tail
        // append leaves every existing mapping intact.
        const auto position = st.derived_to_source.empty() || st.derived_to_source.back() < src_idx
            ? st.derived_to_source.end()
            : std::lower_bound(st.derived_to_source.begin(), st.derived_to_source.end(), src_idx);
        const auto d_idx = static_cast<std::size_t>(position - st.derived_to_source.begin());
        for (auto it = position; it != st.derived_to_source.end(); ++it) ++*it;
        // Insert the new "not yet classified" source slot.
        st.source_to_derived.insert(st.source_to_derived.begin() + static_cast<std::ptrdiff_t>(src_idx),
                                    std::nullopt);
 
        if (!is_in) return;
 
        st.derived_to_source.insert(st.derived_to_source.begin() + static_cast<std::ptrdiff_t>(d_idx),
                                    src_idx);
 
        // The event owns the row even if the source has committed later edits.
        auto shared = ch.item;
        st.items.insert(st.items.begin() + static_cast<std::ptrdiff_t>(d_idx), std::move(shared));
 
        for (std::size_t d = d_idx; d < st.derived_to_source.size(); ++d)
            st.source_to_derived[st.derived_to_source[d]] = d;
 
        lk.unlock();
        sig.emit(ListChange<T>{ListChangeKind::Insert, d_idx, ch.item, 0});
    }
 
    static void handle_remove_(SharedState& st, Signal& 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;   // defensive
 
        std::optional<std::size_t> maybe_d = st.source_to_derived[src_idx];
        st.source_to_derived.erase(st.source_to_derived.begin() + static_cast<std::ptrdiff_t>(src_idx));
 
        // All d2s entries > src_idx shifted by -1 (source collapsed).
        for (auto& s : st.derived_to_source) {
            if (s > src_idx) --s;
        }
 
        if (maybe_d) {
            const std::size_t d_idx = *maybe_d;
            auto removed = st.items[d_idx];   // keep alive past erase
            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));
            renumber_s2d_(st);
            lk.unlock();
            sig.emit(ListChange<T>{ListChangeKind::Remove, d_idx,
                                   removed, 0});
            return;
        }
 
        renumber_s2d_(st);
    }

    static void handle_membership_transition_(SharedState& st, Signal& sig,
                                              const ListChange<T>& ch,
                                              ListChangeKind kind_for_in_in,
                                              bool refresh_value) {
        const bool is_in = (*st.predicate)(*ch.item);
        std::shared_ptr<T> retired;
        std::unique_lock lk(st.m);
        const std::size_t src_idx = ch.index;
 
        if (src_idx >= st.source_to_derived.size()) return;
        const bool was_in = st.source_to_derived[src_idx].has_value();
        if (!was_in && is_in) {
            reserve_one(st.items);
            reserve_one(st.derived_to_source);
        }
 
        if (!was_in && !is_in) return;
 
        if (was_in && is_in) {
            const std::size_t d_idx = *st.source_to_derived[src_idx];
            if (refresh_value) {
                retired = std::exchange(st.items[d_idx], ch.item);
            }
            lk.unlock();
            sig.emit(ListChange<T>{kind_for_in_in, d_idx, ch.item, 0});
            return;
        }
 
        if (was_in && !is_in) {
            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[src_idx] = std::nullopt;
            renumber_s2d_(st);
            lk.unlock();
            sig.emit(ListChange<T>{ListChangeKind::Remove, d_idx,
                                   removed, 0});
            return;
        }
 
        // !was_in && is_in: item became visible.
        std::size_t d_idx = 0;
        for (std::size_t i = 0; i < src_idx; ++i) {
            if (st.source_to_derived[i].has_value()) ++d_idx;
        }
        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), ch.item);
        st.source_to_derived[src_idx] = d_idx;
        renumber_s2d_(st);
        lk.unlock();
        sig.emit(ListChange<T>{ListChangeKind::Insert, d_idx, ch.item, 0});
    }
 
    static void handle_replace_(SharedState& st, Signal& sig,
                                const ListChange<T>& ch) {
        handle_membership_transition_(st, sig, ch,
                                      ListChangeKind::Replace,
                                      /*refresh_value=*/true);
    }
 
    static void handle_item_changed_(SharedState& st, Signal& sig,
                                     const ListChange<T>& ch) {
        handle_membership_transition_(st, sig, ch,
                                      ListChangeKind::ItemChanged,
                                      /*refresh_value=*/false);
    }
 
    static void handle_move_(SharedState& st, Signal& 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() ||
            to   >= st.source_to_derived.size()) {
            return;   // defensive; source mutation out of step
        }
 
        // Derived index of the moved item BEFORE shuffle (if any).
        const std::optional<std::size_t> old_d = st.source_to_derived[from];
 
        // Shuffle source_to_derived to mirror the source's own move.
        auto moved_slot = st.source_to_derived[from];
        st.source_to_derived.erase(st.source_to_derived.begin() + static_cast<std::ptrdiff_t>(from));
        st.source_to_derived.insert(st.source_to_derived.begin() + static_cast<std::ptrdiff_t>(to),
                                    moved_slot);
 
        // If the moved item was in the filter, its derived position
        // needs updating too. Compute the new derived index of the
        // item now at source `to`.
        if (old_d) {
            // Determine new derived index by counting in-filter slots
            // strictly before `to` in the new source layout.
            std::size_t new_d = 0;
            for (std::size_t i = 0; i < to; ++i) {
                if (st.source_to_derived[i].has_value()) ++new_d;
            }
 
            // Physically move the item within derived_to_source and
            // items to reflect the new position.
            if (new_d != *old_d) {
                auto moved_src = st.derived_to_source[*old_d];
                auto moved_item = st.items[*old_d];
                st.derived_to_source.erase(st.derived_to_source.begin() + static_cast<std::ptrdiff_t>(*old_d));
                st.items.erase(st.items.begin() + static_cast<std::ptrdiff_t>(*old_d));
                st.derived_to_source.insert(
                    st.derived_to_source.begin() + static_cast<std::ptrdiff_t>(new_d), moved_src);
                st.items.insert(st.items.begin() + static_cast<std::ptrdiff_t>(new_d), std::move(moved_item));
            }
 
            // Now renumber s2d to reflect both the source shuffle and
            // the derived reordering we just did.
            renumber_s2d_(st);
 
            if (new_d != *old_d) {
                lk.unlock();
                sig.emit(ListChange<T>{ListChangeKind::Move, new_d,
                                       ch.item, *old_d});
            }
            return;
        }
 
        // Moved item was not in the filter — the only change is which
        // source indices each d2s entry refers to.
        renumber_d2s_after_source_move_(st, from, to);
        renumber_s2d_(st);
    }
 
    static void handle_reset_(SharedState& st, Signal& sig, const ListChange<T>& ch) {
        const auto& snapshot = *ch.snapshot;
        std::vector<std::optional<std::size_t>> s2d(snapshot.size(), std::nullopt);
        std::vector<std::size_t> d2s;
        std::vector<std::shared_ptr<T>> items;
        d2s.reserve(snapshot.size());
        items.reserve(snapshot.size());
        for (std::size_t i = 0; i < snapshot.size(); ++i) {
            if (!(*st.predicate)(*snapshot[i])) continue;
            s2d[i] = items.size();
            d2s.push_back(i);
            items.push_back(snapshot[i]);
        }
        auto reset = ListChange<T>::reset(items);
        {
            std::unique_lock lock(st.m);
            st.source_to_derived.swap(s2d);
            st.derived_to_source.swap(d2s);
            st.items.swap(items);
        }
        sig.emit(std::move(reset));
    }
 
    // Walk source_to_derived and assign successive derived indices to
    // every non-nullopt slot. O(source.size()). Call after any
    // structural mutation.
    static void renumber_s2d_(SharedState& st) {
        std::size_t d = 0;
        for (std::size_t s = 0; s < st.source_to_derived.size(); ++s) {
            auto& slot = st.source_to_derived[s];
            if (slot.has_value()) {
                slot = d;
                st.derived_to_source[d++] = s;
            }
        }
    }
 
    // When a filtered-out item moves within the source, we still need
    // to update the source indices stored in derived_to_source so that
    // they refer to the right slots. This is the straightforward
    // translation: values > from and <= to (moving backward) or
    // >= to and < from (moving forward) get shifted by one.
    static void renumber_d2s_after_source_move_(SharedState& st,
                                                std::size_t from,
                                                std::size_t to) {
        if (from < to) {
            for (auto& s : st.derived_to_source) {
                if (s > from && s <= to) --s;
                // NOTE: the moved item itself was NOT in derived,
                // so we never encounter `s == from`.
            }
        } else {
            for (auto& s : st.derived_to_source) {
                if (s >= to && s < from) ++s;
            }
        }
    }
};
 
// ---------------------------------------------------------------------------
//  Factory helper — deduces the source type so pipelines stay readable.
//
//  Without it, chaining forces the caller to spell out every layer:
//
//    auto f = std::make_shared<FilteredList<Task>>(src, pred);
//    auto s = std::make_shared<SortedList<Task, FilteredList<Task>>>(f, cmp);
//
//  With it:
//
//    auto f = aria::filtered(src, pred);
//    auto s = aria::sorted(f, cmp);
//
//  Returns shared_ptr because every derived list takes its source as one, so
//  the result is immediately usable as the next link in the chain.
// ---------------------------------------------------------------------------
template<typename Source,
         typename Predicate,
         typename T = list_source_value_t<Source>>
    requires ListSourceOf<Source, T>
[[nodiscard]] std::shared_ptr<FilteredList<T, Source>>
filtered(std::shared_ptr<Source> source, Predicate predicate) {
    return std::make_shared<FilteredList<T, Source>>(
        std::move(source),
        typename FilteredList<T, Source>::Predicate{std::move(predicate)});
}
 
}  // namespace aria