//! The data seam — the async trait an integrator's row store satisfies.
//!
//! GUI-free and runtime-free: the core defines the shape, produces
//! and consumes futures, and **blocks nowhere**. The integrator drives the
//! futures (iced wraps them in `Task`s and its executor runs them, routing
//! results back as messages).
//!
//! The data source is used as a **generic parameter** (`S: DataSource`), never a
//! trait object, so dispatch is static with no boxing — which is also why the
//! async method can use a native `impl Future` return type instead of
//! `async-trait`. `fetch` returns an owned `Send + 'static` future so the
//! integrator can spawn it on iced's executor (`Task::perform` requires
//! `Future: Send + 'static` on native); resident stores return a ready future.

use core::future::Future;
use core::hash::Hash;
use core::ops::Range;
use std::marker::PhantomData;

use crate::columns::ColumnId;
use crate::query::{FilterValue, QuerySpec, SortDirection};

/// A tolerant row count.
///
/// Resident stores always know their exact count; the held-open out-of-core
/// store may only estimate it, or not know it yet while background indexing runs.
/// The viewport tolerates all three (it just needs *some* total to size the
/// scroll region).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RowCount {
    /// The exact number of rows (resident stores).
    Exact(usize),
    /// A best estimate; the true count may differ (out-of-core).
    Estimated(usize),
    /// Not known yet — e.g. background indexing is still running.
    Unknown,
}

impl RowCount {
    /// The best available count as a `usize`, treating [`Unknown`](Self::Unknown)
    /// as `0`. This is what the viewport uses to size the scroll region.
    #[must_use]
    pub fn lower_bound(&self) -> usize {
        match self {
            RowCount::Exact(n) | RowCount::Estimated(n) => *n,
            RowCount::Unknown => 0,
        }
    }

    /// Whether the count is exact (a resident store).
    #[must_use]
    pub fn is_exact(&self) -> bool {
        matches!(self, RowCount::Exact(_))
    }
}

/// A row store the integrator satisfies — the data seam.
///
/// Async and generic over an opaque [`RowId`](Self::RowId): the core holds and
/// compares row ids (selection, focus, scroll re-anchoring) but never interprets
/// them, so rows survive sort/filter by identity rather than by position.
///
/// ## Imperative transactions and change notification
///
/// Mutation shapes are **store semantics** (an append-only ring, a keyed
/// update-in-place table), which the core never learns — so the seam carries no
/// transaction methods. Stores expose transactions as inherent methods (see
/// [`VecStore`]'s `append`/`drop_front`/`upsert`/`remove`/`clear`), under one
/// seam-level contract: **after any mutation, [`row_count`](Self::row_count),
/// [`fetch`](Self::fetch) and [`index_of`](Self::index_of) reflect the
/// last-applied spec** — the store owns execution, including keeping its
/// derivation consistent.
///
/// The change-notification path is **in-band** in v1: the view owns the store
/// and all mutations route through the view's transactional entry point, which
/// re-queries afterwards (count, re-anchor, window refetch). A push-style
/// subscription is deliberately **not** part of the seam until a store exists
/// that mutates *itself* (the held-open out-of-core store's background
/// indexing) — machinery without a consumer is not built.
///
/// ## Send/'static bounds
///
/// `fetch` returns a `Send + 'static` future and [`Row`](Self::Row) is
/// `Send + 'static` because the integrator spawns fetches on an executor (iced
/// `Task`), which requires owned, sendable futures. These are general async
/// spawning constraints, not toolkit-specific; a single-threaded variant could
/// relax `Send` later via `trait_variant` if a consumer ever needed it.
///
/// ## Implementing a custom (out-of-core) store
///
/// For resident data, reach for [`VecStore`] instead. A custom store is for
/// data behind an async boundary — a database, an IPC channel, a memory-mapped
/// index. The skeleton:
///
/// ```
/// use core::future::Future;
/// use std::ops::Range;
///
/// use oriel_core::data::{DataSource, RowCount};
/// use oriel_core::query::QuerySpec;
///
/// #[derive(Clone)]
/// struct Record {
///     key: u64,
///     payload: String,
/// }
///
/// struct RemoteStore {
///     /// The last-applied spec — `fetch` must execute against it.
///     spec: QuerySpec,
/// }
///
/// impl DataSource for RemoteStore {
///     type RowId = u64;
///     type Row = Record;
///
///     fn row_count(&self) -> RowCount {
///         // An estimate is fine — the viewport only needs *some* total,
///         // and `Unknown` is legal while background indexing runs.
///         RowCount::Estimated(1_000_000)
///     }
///
///     fn fetch(&self, range: Range<usize>) -> impl Future<Output = Vec<Record>> + Send + 'static {
///         let spec = self.spec.clone();
///         async move {
///             // Issue the real query for `range` under `spec` here; the
///             // grid renders placeholder rows until this resolves and
///             // discards results superseded by a newer fetch.
///             let _ = (range, spec);
///             Vec::new()
///         }
///     }
///
///     fn row_id(&self, row: &Record) -> u64 {
///         row.key
///     }
///
///     fn apply_spec(&mut self, spec: &QuerySpec) {
///         // Record now, execute behind `fetch` — this call must stay cheap.
///         self.spec = spec.clone();
///     }
/// }
/// ```
///
/// Override [`index_of`](Self::index_of) whenever the store can answer
/// "where is this row now?" in reasonable time — it powers scroll
/// re-anchoring across mutations; the tolerant default (`None`) degrades to
/// positional anchoring.
pub trait DataSource {
    /// Opaque row identity, compared but never interpreted by the core.
    type RowId: Eq + Hash + Clone;

    /// The row payload a fetch yields — owned and sendable so it can cross the
    /// executor boundary as a `Task` result.
    type Row: Send + 'static;

    /// The current row count (tolerant — see [`RowCount`]).
    fn row_count(&self) -> RowCount;

    /// Fetches the rows for the half-open index `range`, in the current order.
    ///
    /// Returns a `Send + 'static` future resolving to the rows (owned). Resident
    /// stores resolve immediately; the held-open out-of-core store may take time,
    /// during which the integrator renders placeholder rows and fills them when
    /// the future resolves. A store may return **fewer** rows than requested if
    /// the range runs past the (tolerant) end of the data.
    ///
    /// Expected traffic from the shipped grid: windowed ranges of a few
    /// hundred rows around the viewport, re-requested on scroll near the
    /// window edge, after every transaction, and on spec changes — with
    /// stale results discarded integrator-side. See the grid's module docs
    /// ("Fetch traffic") for the precise policy.
    fn fetch(
        &self,
        range: Range<usize>,
    ) -> impl Future<Output = Vec<Self::Row>> + Send + 'static;

    /// The [`RowId`](Self::RowId) of a row — cheap and pure.
    fn row_id(&self, row: &Self::Row) -> Self::RowId;

    /// Best-effort position of the row with `id` in the **current** order
    /// (post-spec), or `None` when the row is absent, filtered out, or the
    /// store cannot answer cheaply (the default).
    ///
    /// This is the re-anchoring primitive (rows appended/removed
    /// above the viewport must not make visible content jump — re-anchor to a
    /// stable row by `RowId`): after a transaction, the view asks where its
    /// anchor row went and restores the scroll position around it. Callers
    /// must degrade gracefully on `None` (fall back to a positional anchor);
    /// stores should override this when they can answer in reasonable time.
    fn index_of(&self, _id: &Self::RowId) -> Option<usize> {
        None
    }

    /// Applies a sort/filter spec; subsequent [`row_count`](Self::row_count) and
    /// [`fetch`](Self::fetch) calls reflect it.
    ///
    /// This call itself is **synchronous and cheap by contract** — it records
    /// (and for resident stores, executes) the spec. An out-of-core store
    /// records the spec here and executes it *behind* its async `fetch`/tolerant
    /// `row_count` (returning `Estimated`/`Unknown` and pending fetches while
    /// the pushed-down query runs), so async execution needs no extra trait
    /// surface. The straddle: the caller owns the spec (view state); the store
    /// owns execution.
    fn apply_spec(&mut self, spec: &QuerySpec);
}

/// A resident, in-memory [`DataSource`] backed by a `Vec`.
///
/// A **public example** of the trait — no privileged internals: it holds all
/// rows and returns ready futures. `key_of` derives each row's
/// [`RowId`](DataSource::RowId).
///
/// Sort/filter specs execute in memory over an **index vector** (a `Vec` of
/// positions into the backing rows): filtering selects indices,
/// sorting reorders them, and the data itself is never reordered or cloned.
/// Executing specs requires a field resolver ([`with_fields`]); without one,
/// every field is unresolvable, so a non-empty filter matches nothing (loud)
/// and a sort leaves the order unchanged (all keys tie).
///
/// Its `fetch` clones the requested slice up front, so the returned future owns
/// its data and is `Send + 'static` (it borrows nothing from `self`).
///
/// # Cost model (for high-rate ingest)
///
/// With **no spec applied**, mutations are plain `Vec` operations. With a
/// sort/filter **retained**, every mutation re-derives the index vector to
/// uphold the seam's post-mutation consistency contract — O(n) filter plus
/// O(n log n) sort per transaction. At ingest rates where that matters,
/// batch mutations into one transaction per tick (the examples ingest at
/// 60 Hz over tens of thousands of rows this way) or drop the sort during
/// heavy ingest.
///
/// # Example
///
/// ```
/// use oriel_core::columns::ColumnId;
/// use oriel_core::data::{DataSource, VecStore};
/// use oriel_core::query::{FilterValue, QuerySpec, SortDirection, SortSpec};
///
/// #[derive(Clone)]
/// struct Person {
///     id: u64,
///     name: String,
/// }
///
/// const NAME_COL: ColumnId = ColumnId(1);
///
/// fn key(p: &Person) -> u64 {
///     p.id
/// }
///
/// let people = vec![
///     Person { id: 1, name: "Ada".into() },
///     Person { id: 2, name: "Grace".into() },
/// ];
/// // The resolver makes sort/filter specs executable against NAME_COL.
/// let mut store = VecStore::new(people, key as fn(&Person) -> u64)
///     .with_fields(|p, col| match col {
///         NAME_COL => Some(FilterValue::Text(p.name.clone())),
///         _ => None,
///     });
///
/// // Mutations are inherent store methods (append/upsert/remove/…); in an
/// // app they run inside the grid's `transact` so the view reacts.
/// store.append(vec![Person { id: 3, name: "Edsger".into() }]);
///
/// // The store executes specs; the spec itself is plain data.
/// store.apply_spec(&QuerySpec {
///     sort: Some(SortSpec { column: NAME_COL, direction: SortDirection::Descending }),
///     filter: Default::default(),
/// });
/// assert_eq!(store.row_count().lower_bound(), 3);
/// assert_eq!(store.index_of(&2), Some(0)); // "Grace" sorts first, descending
/// ```
///
/// [`with_fields`]: Self::with_fields
pub struct VecStore<Row, Id, KeyOf> {
    rows: Vec<Row>,
    key_of: KeyOf,
    /// Resolves `(row, column)` to a field value for spec execution.
    fields: Option<Box<dyn Fn(&Row, ColumnId) -> Option<FilterValue>>>,
    /// The last-applied spec — retained so transactions can re-derive without
    /// the caller re-applying it (post-mutation consistency is the store's
    /// obligation — see the [`DataSource`] transaction contract).
    spec: QuerySpec,
    /// Positions into `rows` in the current query order; `None` = identity
    /// (no spec applied — the fast path).
    indices: Option<Vec<usize>>,
    // `fn() -> Id` keeps `VecStore` `Send`/`Sync` regardless of `Id` and marks
    // `Id` as produced-not-stored.
    _id: PhantomData<fn() -> Id>,
}

impl<Row, Id, KeyOf> VecStore<Row, Id, KeyOf>
where
    Id: Eq + Hash + Clone,
    KeyOf: Fn(&Row) -> Id,
{
    /// Creates a resident store over `rows`, deriving row ids with `key_of`.
    pub fn new(rows: Vec<Row>, key_of: KeyOf) -> Self {
        Self {
            rows,
            key_of,
            fields: None,
            spec: QuerySpec::default(),
            indices: None,
            _id: PhantomData,
        }
    }

    /// Attaches the field resolver used to execute sort/filter specs: given a
    /// row and a [`ColumnId`], return that field's value (or `None` if the
    /// column has no value for this row).
    #[must_use]
    pub fn with_fields(
        mut self,
        fields: impl Fn(&Row, ColumnId) -> Option<FilterValue> + 'static,
    ) -> Self {
        self.fields = Some(Box::new(fields));
        self
    }

    /// The backing rows, in **storage** order (not the current query order).
    #[must_use]
    pub fn rows(&self) -> &[Row] {
        &self.rows
    }

    // --- imperative transactions ---
    //
    // Inherent methods, not trait surface: mutation shapes are store semantics
    // the core never learns. Every transaction ends by re-deriving the index
    // vector against the retained spec (recompute-from-scratch — the
    // accepted default; make it incremental only if profiling real 20k-row
    // ingest demands it), which is what keeps the DataSource post-mutation
    // consistency contract.

    /// Appends `rows` at the end — the append-stream hot path. Batch
    /// the appends of one ingest tick into a single call: the re-derivation
    /// runs once per call, not once per row.
    pub fn append(&mut self, rows: impl IntoIterator<Item = Row>) {
        self.rows.extend(rows);
        self.rederive();
    }

    /// Removes the first `n` rows (clamped) — a ring's over-cap drop.
    pub fn drop_front(&mut self, n: usize) {
        let n = n.min(self.rows.len());
        drop(self.rows.drain(..n));
        self.rederive();
    }

    /// Inserts `row` at storage position `index` (clamped to the end). Note
    /// that the *view* position is governed by the applied spec: under a
    /// sort, storage position is irrelevant (`append`/`upsert` suffice) —
    /// this matters when the natural order IS the view order.
    pub fn insert(&mut self, index: usize, row: Row) {
        let index = index.min(self.rows.len());
        self.rows.insert(index, row);
        self.rederive();
    }

    /// Keeps only the rows for which `keep` returns `true` — bulk arbitrary
    /// deletion with a single re-derivation.
    pub fn retain(&mut self, keep: impl FnMut(&Row) -> bool) {
        self.rows.retain(keep);
        self.rederive();
    }

    /// Replaces the row sharing `row`'s id in place, or appends `row` if no
    /// such row exists — the keyed live-table update path. Returns
    /// `true` if an existing row was replaced.
    pub fn upsert(&mut self, row: Row) -> bool {
        let id = (self.key_of)(&row);
        let key_of = &self.key_of;
        let replaced = match self.rows.iter_mut().find(|r| key_of(r) == id) {
            Some(slot) => {
                *slot = row;
                true
            }
            None => {
                self.rows.push(row);
                false
            }
        };
        self.rederive();
        replaced
    }

    /// Removes the row with `id`, if present. Returns `true` if it existed.
    pub fn remove(&mut self, id: &Id) -> bool {
        let key_of = &self.key_of;
        match self.rows.iter().position(|r| key_of(r) == *id) {
            Some(i) => {
                self.rows.remove(i);
                self.rederive();
                true
            }
            None => false,
        }
    }

    /// Removes all rows — the wholesale reset.
    pub fn clear(&mut self) {
        self.rows.clear();
        self.rederive();
    }

    /// Recomputes the index vector from the retained spec.
    fn rederive(&mut self) {
        if self.spec.is_identity() {
            self.indices = None;
            return;
        }
        let fields = self.fields.as_deref();
        let resolve = |row: &Row, col: ColumnId| fields.and_then(|f| f(row, col));

        let indices = self
            .rows
            .iter()
            .enumerate()
            .filter(|(_, row)| self.spec.filter.matches(|col| resolve(row, col)))
            .map(|(i, _)| i);

        self.indices = Some(match &self.spec.sort {
            None => indices.collect(),
            Some(sort) => {
                // Precompute each survivor's key once, then stable-sort.
                let mut keyed: Vec<(Option<FilterValue>, usize)> = indices
                    .map(|i| (resolve(&self.rows[i], sort.column), i))
                    .collect();
                keyed.sort_by(|(a, _), (b, _)| compare_keys(a, b, sort.direction));
                keyed.into_iter().map(|(_, i)| i).collect()
            }
        });
    }
}

/// Ordering of two sort keys: resolvable keys order by [`FilterValue::compare`]
/// (reversed when descending); rows whose key is unresolvable sort last
/// regardless of direction, so they never pollute the top of the view.
fn compare_keys(
    a: &Option<FilterValue>,
    b: &Option<FilterValue>,
    direction: SortDirection,
) -> core::cmp::Ordering {
    use core::cmp::Ordering;
    match (a, b) {
        (Some(a), Some(b)) => {
            let ord = a.compare(b);
            match direction {
                SortDirection::Ascending => ord,
                SortDirection::Descending => ord.reverse(),
            }
        }
        (Some(_), None) => Ordering::Less,
        (None, Some(_)) => Ordering::Greater,
        (None, None) => Ordering::Equal,
    }
}

impl<Row, Id, KeyOf> DataSource for VecStore<Row, Id, KeyOf>
where
    Row: Clone + Send + 'static,
    Id: Eq + Hash + Clone,
    KeyOf: Fn(&Row) -> Id,
{
    type RowId = Id;
    type Row = Row;

    fn row_count(&self) -> RowCount {
        RowCount::Exact(self.indices.as_ref().map_or(self.rows.len(), Vec::len))
    }

    fn fetch(&self, range: Range<usize>) -> impl Future<Output = Vec<Row>> + Send + 'static {
        let rows = match &self.indices {
            None => {
                let start = range.start.min(self.rows.len());
                let end = range.end.min(self.rows.len()).max(start);
                self.rows[start..end].to_vec()
            }
            Some(indices) => {
                let start = range.start.min(indices.len());
                let end = range.end.min(indices.len()).max(start);
                indices[start..end]
                    .iter()
                    .map(|&i| self.rows[i].clone())
                    .collect()
            }
        };
        core::future::ready(rows)
    }

    fn row_id(&self, row: &Row) -> Id {
        (self.key_of)(row)
    }

    /// Exact answer by linear scan of the current order — fine for resident
    /// sizes (a few index_of calls per transaction over ≤ tens of thousands of
    /// rows); a keyed or sorted store can do better.
    fn index_of(&self, id: &Id) -> Option<usize> {
        let key_of = &self.key_of;
        match &self.indices {
            None => self.rows.iter().position(|r| key_of(r) == *id),
            Some(indices) => indices.iter().position(|&i| key_of(&self.rows[i]) == *id),
        }
    }

    fn apply_spec(&mut self, spec: &QuerySpec) {
        self.spec = spec.clone();
        self.rederive();
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Drives a future that is expected to already be ready — the resident
    /// store's contract. No runtime, no dependencies (`Waker::noop` is std),
    /// so `oriel-core` stays dependency-free even while testing the async seam.
    fn block_on_ready<F: Future>(future: F) -> F::Output {
        use core::task::{Context, Poll};
        let mut fut = core::pin::pin!(future);
        let waker = core::task::Waker::noop();
        let mut cx = Context::from_waker(waker);
        match fut.as_mut().poll(&mut cx) {
            Poll::Ready(v) => v,
            Poll::Pending => panic!("a resident DataSource future must be ready"),
        }
    }

    fn demo_store() -> VecStore<(u64, &'static str), u64, fn(&(u64, &'static str)) -> u64> {
        let rows = (0..5u64).map(|i| (i, "row")).collect();
        VecStore::new(rows, |&(id, _)| id)
    }

    #[test]
    fn vecstore_reports_exact_count() {
        let store = demo_store();
        assert_eq!(store.row_count(), RowCount::Exact(5));
        assert!(store.row_count().is_exact());
        assert_eq!(store.row_count().lower_bound(), 5);
    }

    #[test]
    fn vecstore_fetch_returns_requested_slice() {
        let store = demo_store();
        let rows = block_on_ready(store.fetch(1..3));
        assert_eq!(rows, vec![(1, "row"), (2, "row")]);
    }

    #[test]
    fn vecstore_fetch_clamps_past_the_end() {
        let store = demo_store();
        // Range past the end yields only the tail, never panics.
        assert_eq!(block_on_ready(store.fetch(3..100)).len(), 2);
        assert_eq!(block_on_ready(store.fetch(10..20)).len(), 0);
        // Inverted range yields empty rather than panicking.
        assert_eq!(block_on_ready(store.fetch(4..2)).len(), 0);
    }

    #[test]
    fn vecstore_row_id_uses_key_of() {
        let store = demo_store();
        assert_eq!(store.row_id(&(42, "x")), 42);
    }

    // --- A second, unrelated store impl: proves the
    // seam is implementable by an arbitrary type, with a *non-Copy* `RowId`. ---

    #[derive(Clone, Debug, PartialEq)]
    struct Metric {
        key: String,
        count: u64,
    }

    struct StatsStore {
        rows: Vec<Metric>,
    }

    impl DataSource for StatsStore {
        type RowId = String; // deliberately non-Copy
        type Row = Metric;

        fn row_count(&self) -> RowCount {
            RowCount::Exact(self.rows.len())
        }

        fn fetch(&self, range: Range<usize>) -> impl Future<Output = Vec<Metric>> + Send + 'static {
            let start = range.start.min(self.rows.len());
            let end = range.end.min(self.rows.len()).max(start);
            let rows = self.rows[start..end].to_vec();
            core::future::ready(rows)
        }

        fn row_id(&self, row: &Metric) -> String {
            row.key.clone()
        }

        // This double exists to prove the seam's *shape*; it deliberately
        // supports no reordering, so recording the spec is a no-op.
        fn apply_spec(&mut self, _spec: &QuerySpec) {}
    }

    /// Generic over *any* `DataSource` — the shape the integrator's widget code
    /// takes, so it works against every store with no change.
    fn total_rows<S: DataSource>(store: &S) -> usize {
        store.row_count().lower_bound()
    }

    #[test]
    fn a_second_store_satisfies_the_same_trait() {
        let store = StatsStore {
            rows: vec![
                Metric { key: "0:7DF".into(), count: 12 },
                Metric { key: "1:7E8".into(), count: 3 },
            ],
        };
        // Same generic entry points as the Vec store — no per-store code.
        assert_eq!(total_rows(&store), 2);
        let rows = block_on_ready(store.fetch(0..2));
        assert_eq!(rows.len(), 2);
        assert_eq!(store.row_id(&rows[0]), "0:7DF".to_string());

        // And the generic helper works identically over the Vec store.
        assert_eq!(total_rows(&demo_store()), 5);
    }

    // --- sort + filter on the index vector, executed by the store ---

    use crate::query::{FilterSpec, Predicate, SortSpec};

    const CH_COL: ColumnId = ColumnId(1);
    const TEXT_COL: ColumnId = ColumnId(2);
    const COUNT_COL: ColumnId = ColumnId(3);

    /// A Trace-like frame: channel + message text + a per-key count.
    #[derive(Clone, Debug, PartialEq)]
    struct Frame {
        ch: i64,
        text: &'static str,
        count: i64,
    }

    fn frame_store() -> VecStore<Frame, &'static str, fn(&Frame) -> &'static str> {
        let rows = vec![
            Frame { ch: 0, text: "ID 7DF DLC 8", count: 12 },
            Frame { ch: 1, text: "ID 7E8 DLC 8", count: 40 },
            Frame { ch: 0, text: "ID 7E0 DLC 8", count: 7 },
            Frame { ch: 1, text: "ID 7DF DLC 3", count: 40 },
            Frame { ch: 0, text: "ID 123 DLC 8", count: 99 },
        ];
        fn key(f: &Frame) -> &'static str {
            f.text // unique per fixture row
        }
        VecStore::new(rows, key as fn(&Frame) -> &'static str).with_fields(|f, col| match col {
            CH_COL => Some(FilterValue::Int(f.ch)),
            TEXT_COL => Some(FilterValue::Text(f.text.to_string())),
            COUNT_COL => Some(FilterValue::Int(f.count)),
            _ => None,
        })
    }

    /// A representable predicate filter (search text + channel), executed by
    /// the store on the index vector.
    #[test]
    fn trace_like_predicate_filter() {
        let mut store = frame_store();
        store.apply_spec(&QuerySpec {
            sort: None,
            filter: FilterSpec {
                clauses: vec![
                    Predicate::Contains { column: TEXT_COL, needle: "7df".into() },
                    Predicate::Equals { column: CH_COL, value: FilterValue::Int(0) },
                ],
            },
        });
        assert_eq!(store.row_count(), RowCount::Exact(1));
        let rows = block_on_ready(store.fetch(0..10));
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].text, "ID 7DF DLC 8");
        // The backing data is untouched — only the index vector changed.
        assert_eq!(store.rows().len(), 5);
    }

    /// Sort a column descending (stable for ties).
    #[test]
    fn stats_count_desc_sort() {
        let mut store = frame_store();
        store.apply_spec(&QuerySpec {
            sort: Some(SortSpec { column: COUNT_COL, direction: SortDirection::Descending }),
            filter: FilterSpec::none(),
        });
        let counts: Vec<i64> =
            block_on_ready(store.fetch(0..10)).iter().map(|f| f.count).collect();
        assert_eq!(counts, vec![99, 40, 40, 12, 7]);
    }

    #[test]
    fn filter_then_sort_compose() {
        let mut store = frame_store();
        store.apply_spec(&QuerySpec {
            sort: Some(SortSpec { column: COUNT_COL, direction: SortDirection::Ascending }),
            filter: FilterSpec {
                clauses: vec![Predicate::Equals { column: CH_COL, value: FilterValue::Int(0) }],
            },
        });
        let counts: Vec<i64> =
            block_on_ready(store.fetch(0..10)).iter().map(|f| f.count).collect();
        assert_eq!(counts, vec![7, 12, 99]);
    }

    #[test]
    fn fetch_windows_map_through_the_index_vector() {
        let mut store = frame_store();
        store.apply_spec(&QuerySpec {
            sort: Some(SortSpec { column: COUNT_COL, direction: SortDirection::Descending }),
            filter: FilterSpec::none(),
        });
        // A mid-window fetch in sorted order: [99, 40, 40, 12, 7][1..3] = [40, 40].
        let window = block_on_ready(store.fetch(1..3));
        assert_eq!(window.iter().map(|f| f.count).collect::<Vec<_>>(), vec![40, 40]);
        // Past-the-end clamps against the *filtered* length, not the backing length.
        assert_eq!(block_on_ready(store.fetch(4..99)).len(), 1);
    }

    #[test]
    fn identity_spec_restores_natural_order() {
        let mut store = frame_store();
        store.apply_spec(&QuerySpec {
            sort: Some(SortSpec { column: COUNT_COL, direction: SortDirection::Descending }),
            filter: FilterSpec {
                clauses: vec![Predicate::Equals { column: CH_COL, value: FilterValue::Int(1) }],
            },
        });
        assert_eq!(store.row_count(), RowCount::Exact(2));
        store.apply_spec(&QuerySpec::default());
        assert_eq!(store.row_count(), RowCount::Exact(5));
        assert_eq!(block_on_ready(store.fetch(0..1))[0].count, 12);
    }

    // --- imperative transactions + index_of ---

    /// `index_of` answers in the **current** order: natural, sorted, and
    /// filtered-out/absent both yield `None`-or-position correctly.
    #[test]
    fn index_of_reflects_current_order() {
        let mut store = frame_store();
        // Natural order: "ID 7E0 DLC 8" is storage position 2.
        assert_eq!(store.index_of(&"ID 7E0 DLC 8"), Some(2));

        // COUNT descending: counts [99, 40, 40, 12, 7] — count-7 row is last.
        store.apply_spec(&QuerySpec {
            sort: Some(SortSpec { column: COUNT_COL, direction: SortDirection::Descending }),
            filter: FilterSpec::none(),
        });
        assert_eq!(store.index_of(&"ID 7E0 DLC 8"), Some(4));

        // Filtered to ch 1: the ch-0 row is filtered out -> None; absent -> None.
        store.apply_spec(&QuerySpec {
            sort: None,
            filter: FilterSpec {
                clauses: vec![Predicate::Equals { column: CH_COL, value: FilterValue::Int(1) }],
            },
        });
        assert_eq!(store.index_of(&"ID 7E0 DLC 8"), None);
        assert_eq!(store.index_of(&"ID 7E8 DLC 8"), Some(0));
        assert_eq!(store.index_of(&"no such row"), None);
    }

    /// The trait default is tolerant `None` — a store that can't answer cheaply
    /// simply doesn't override, and callers must degrade gracefully.
    #[test]
    fn default_index_of_is_tolerant_none() {
        let store = StatsStore {
            rows: vec![Metric { key: "0:7DF".into(), count: 12 }],
        };
        assert_eq!(store.index_of(&"0:7DF".to_string()), None);
    }

    /// THE store contract: after transactions, `row_count`/`fetch`/`index_of`
    /// reflect the last-applied spec **without** the caller re-applying it.
    #[test]
    fn transactions_keep_derivation_consistent_without_reapply() {
        let mut store = frame_store();
        store.apply_spec(&QuerySpec {
            sort: Some(SortSpec { column: COUNT_COL, direction: SortDirection::Descending }),
            filter: FilterSpec {
                clauses: vec![Predicate::Equals { column: CH_COL, value: FilterValue::Int(0) }],
            },
        });
        // ch-0 rows sorted by count desc: [99, 12, 7].
        assert_eq!(store.row_count(), RowCount::Exact(3));

        // Append one matching (ch 0, count 50) and one non-matching (ch 1) row.
        store.append([
            Frame { ch: 0, text: "ID 456 DLC 8", count: 50 },
            Frame { ch: 1, text: "ID 789 DLC 8", count: 1000 },
        ]);
        assert_eq!(store.row_count(), RowCount::Exact(4), "only the ch-0 row joins the view");
        let counts: Vec<i64> =
            block_on_ready(store.fetch(0..10)).iter().map(|f| f.count).collect();
        assert_eq!(counts, vec![99, 50, 12, 7], "new row sorted into place");
        assert_eq!(store.index_of(&"ID 456 DLC 8"), Some(1));

        // Ring drop from the front of STORAGE order (drops "ID 7DF DLC 8",
        // count 12, ch 0 — a row currently visible in the derived view).
        store.drop_front(1);
        assert_eq!(store.row_count(), RowCount::Exact(3));
        let counts: Vec<i64> =
            block_on_ready(store.fetch(0..10)).iter().map(|f| f.count).collect();
        assert_eq!(counts, vec![99, 50, 7]);
        assert_eq!(store.index_of(&"ID 7DF DLC 8"), None, "dropped row is gone from the view");
    }

    /// Mode B update-in-place: `upsert` replaces by id (count stable, values
    /// re-derived) or appends when the id is new.
    #[test]
    fn upsert_updates_in_place_or_appends() {
        let mut store = frame_store();
        store.apply_spec(&QuerySpec {
            sort: Some(SortSpec { column: COUNT_COL, direction: SortDirection::Descending }),
            filter: FilterSpec::none(),
        });

        // Existing id, new count: replaced in place and re-sorted to the top.
        let replaced = store.upsert(Frame { ch: 0, text: "ID 7E0 DLC 8", count: 500 });
        assert!(replaced);
        assert_eq!(store.rows().len(), 5, "in-place: storage count unchanged");
        assert_eq!(store.index_of(&"ID 7E0 DLC 8"), Some(0), "re-sorted by new count");

        // New id: appended.
        let replaced = store.upsert(Frame { ch: 2, text: "ID NEW DLC 8", count: 1 });
        assert!(!replaced);
        assert_eq!(store.row_count(), RowCount::Exact(6));
        assert_eq!(store.index_of(&"ID NEW DLC 8"), Some(5), "count 1 sorts last");
    }

    /// Arbitrary positional insertion and bulk deletion keep the derivation
    /// consistent like every other transaction — and under a sort, the view
    /// position comes from the spec, not the storage position.
    #[test]
    fn insert_and_retain_keep_derivation_consistent() {
        let mut store = frame_store();
        store.apply_spec(&QuerySpec {
            sort: Some(SortSpec { column: COUNT_COL, direction: SortDirection::Descending }),
            filter: FilterSpec::none(),
        });

        // Storage position honored; view position governed by the sort.
        store.insert(2, Frame { ch: 0, text: "ID 999 DLC 8", count: 55 });
        assert_eq!(store.rows()[2].text, "ID 999 DLC 8");
        assert_eq!(store.index_of(&"ID 999 DLC 8"), Some(1), "55 sorts between 99 and 40");

        // Bulk deletion, one re-derivation: view keeps [99, 55, 40, 40].
        store.retain(|f| f.count >= 40);
        assert_eq!(store.row_count(), RowCount::Exact(4));
        let counts: Vec<i64> =
            block_on_ready(store.fetch(0..10)).iter().map(|f| f.count).collect();
        assert_eq!(counts, vec![99, 55, 40, 40]);

        // Past-the-end insert clamps to append.
        store.insert(999, Frame { ch: 1, text: "ID TAIL", count: 1 });
        assert_eq!(store.rows().last().unwrap().text, "ID TAIL");
    }

    #[test]
    fn remove_and_clear() {
        let mut store = frame_store();
        assert!(store.remove(&"ID 7E8 DLC 8"));
        assert!(!store.remove(&"ID 7E8 DLC 8"), "second remove finds nothing");
        assert_eq!(store.row_count(), RowCount::Exact(4));

        store.clear();
        assert_eq!(store.row_count(), RowCount::Exact(0));
        assert_eq!(block_on_ready(store.fetch(0..10)).len(), 0);
    }

    /// Without a field resolver the store cannot evaluate predicates: a
    /// non-empty filter matches nothing (loud, not silently ignored) and a sort
    /// leaves the order unchanged (every key ties).
    #[test]
    fn store_without_fields_is_loud_not_silent() {
        let mut store = demo_store(); // no .with_fields
        store.apply_spec(&QuerySpec {
            sort: None,
            filter: FilterSpec {
                clauses: vec![Predicate::Contains { column: TEXT_COL, needle: "row".into() }],
            },
        });
        assert_eq!(store.row_count(), RowCount::Exact(0));

        store.apply_spec(&QuerySpec {
            sort: Some(SortSpec { column: TEXT_COL, direction: SortDirection::Descending }),
            filter: FilterSpec::none(),
        });
        let ids: Vec<u64> = block_on_ready(store.fetch(0..5)).iter().map(|r| r.0).collect();
        assert_eq!(ids, vec![0, 1, 2, 3, 4], "unresolvable keys all tie; stable order kept");
    }
}
