hayson-engine 0.22.0

use std::collections::{HashMap, HashSet};
use std::fmt;
use std::io;

use serde_json::Value;

use crate::errors::HaysonCipherError;
use crate::registry::composite_index::{RegistryIndex, RegistryIndexKind};
use crate::registry::exact_index::value_key;
use crate::registry::fulltext_index::SearchMode;
use crate::registry::sorted_index::{RangeBounds, SortValue};
use crate::registry::table_registry::TableRegistry;
use crate::schema::{HaysonRecord, UNPARTITIONED};
use crate::storage::record::parse_entry;
use crate::storage::{LogFile, LogSet, Slot};

use super::predicates::{all_leaves, leaves_of, same_value, test_node, ComparisonOp, Predicate, PredicateNode};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QueryStrategy {
    Primary,
    Exact,
    Sorted,
    Fulltext,
    Union,
    Scan,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PlanBranch {
    pub strategy: QueryStrategy,
    pub field: Option<String>,
    pub candidates: usize,
}

#[derive(Debug, Clone, PartialEq)]
pub struct AccessPath {
    pub strategy: QueryStrategy,
    pub field: Option<String>,
    pub index: Option<String>,
    pub keys: Vec<String>,
    pub branches: Option<Vec<PlanBranch>>,
    /// Relevance of every key a `spec.search` scored, kept alongside the candidates because the
    /// ranking it drives is applied after the predicate filter, not while planning.
    pub scores: Option<HashMap<String, f64>>,
    /// Set only when the query ran through a unit of work's overlay — see [`PlanOverlay`].
    pub overlay: Option<PlanOverlay>,
}

/// How the staged candidates of a [`PlanOverlay`] were picked.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OverlayStrategy {
    /// The overlay's own exact index answered a top-level equality.
    Exact,
    /// The staged records were filtered by the search text, through the rescored corpus.
    Fulltext,
    /// No overlay index could narrow the predicate, so every staged record was checked.
    Scan,
}

/// What a unit of work's staged writes did to a query plan.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PlanOverlay {
    /// Staged records that joined the candidate set.
    pub staged: usize,
    /// Committed candidates the overlay took back out — re-staged in this unit, or staged deleted.
    pub suppressed: usize,
    /// The BM25 statistics behind the candidates were recomputed over the overlaid corpus.
    pub rescored: bool,
    pub strategy: OverlayStrategy,
    /// Records the unit holds for this table, whether or not they could match.
    pub size: usize,
}

/// Candidate keys an overlay's own index can prove for one field/value pair.
pub enum OverlayCandidates {
    /// The overlay indexes nothing on that field, so it cannot narrow and every staged record stands.
    NotIndexed,
    /// The overlay indexes the field; these are the staged keys holding the value, possibly none.
    Narrowed(Vec<String>),
}

/// The unit-of-work overlay a query is planned through — the `txn/` side of the extension point
/// `RecordSource` names. Kept as a trait here so `query/` stays below `txn/` in the layering.
pub trait QueryOverlay {
    /// Every key the unit stages a value for, in the state the unit will leave it.
    fn records(&self) -> &HashMap<String, HaysonRecord>;
    /// Every key the unit stages a delete for.
    fn deleted(&self) -> &HashSet<String>;
    fn candidates_for(&self, field: &str, encoded: &str) -> OverlayCandidates;
}

#[derive(Debug, Clone, PartialEq)]
pub struct SearchSpec {
    pub text: String,
    pub mode: SearchMode,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SortDirection {
    Asc,
    Desc,
}

/// The single sort key of a query. Single, not a list, because the TS `QueryBuilder.orderBy`
/// it mirrors keeps one key and overwrites it on every call.
#[derive(Debug, Clone, PartialEq)]
pub struct OrderBy {
    pub field: String,
    pub direction: SortDirection,
}

#[derive(Debug, Clone, PartialEq)]
pub struct QuerySpec {
    pub root: PredicateNode,
    pub search: Option<SearchSpec>,
    pub order_by: Option<OrderBy>,
    pub limit: Option<usize>,
    pub offset: Option<usize>,
    /// The caller wants `matched` and the plan, not the page: `QueryResult::keys` comes back empty
    /// and no page is ordered, so a binding has nothing to load records for. What the TS
    /// `QueryBuilder.count()`/`.explain()` ask for.
    pub count_only: bool,
}

impl QuerySpec {
    pub fn new(root: PredicateNode) -> Self {
        QuerySpec {
            root,
            search: None,
            order_by: None,
            limit: None,
            offset: None,
            count_only: false,
        }
    }

    pub fn with_search(mut self, text: impl Into<String>, mode: SearchMode) -> Self {
        self.search = Some(SearchSpec {
            text: text.into(),
            mode,
        });
        self
    }

    pub fn with_order_by(mut self, field: impl Into<String>, direction: SortDirection) -> Self {
        self.order_by = Some(OrderBy {
            field: field.into(),
            direction,
        });
        self
    }

    pub fn with_limit(mut self, limit: usize) -> Self {
        self.limit = Some(limit);
        self
    }

    pub fn with_offset(mut self, offset: usize) -> Self {
        self.offset = Some(offset);
        self
    }

    pub fn counting(mut self, count_only: bool) -> Self {
        self.count_only = count_only;
        self
    }

    /// The same spec with the page window dropped — what an aggregate runs, since `limit`/`offset`
    /// name a window over the answer, not over the rows an aggregate folds. `count_only` is dropped
    /// with it: an aggregate folds the matched keys one record at a time, so it needs them listed.
    pub fn unpaged(&self) -> Self {
        QuerySpec {
            limit: None,
            offset: None,
            count_only: false,
            ..self.clone()
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct KeySet {
    strategy: QueryStrategy,
    field: Option<String>,
    index: Option<String>,
    keys: Vec<String>,
    branches: Option<Vec<PlanBranch>>,
}

impl KeySet {
    fn into_path(self, scores: Option<HashMap<String, f64>>) -> AccessPath {
        AccessPath {
            strategy: self.strategy,
            field: self.field,
            index: self.index,
            keys: self.keys,
            branches: self.branches,
            scores,
            overlay: None,
        }
    }
}

fn cheaper(best: Option<KeySet>, candidate: Option<KeySet>) -> Option<KeySet> {
    match candidate {
        None => best,
        Some(candidate) => match best {
            None => Some(candidate),
            Some(best) => {
                if candidate.keys.len() < best.keys.len() {
                    Some(candidate)
                } else {
                    Some(best)
                }
            }
        },
    }
}

fn is_primitive(value: &Value) -> bool {
    matches!(value, Value::String(_) | Value::Number(_) | Value::Bool(_))
}

fn is_primitive_equality(predicate: &Predicate) -> bool {
    predicate.op == ComparisonOp::Eq && is_primitive(&predicate.value)
}

fn equality_on<'a>(leaves: &[&'a Predicate], field: &str) -> Option<&'a Predicate> {
    let usable: Vec<&'a Predicate> = leaves
        .iter()
        .copied()
        .filter(|leaf| {
            leaf.field == field
                && (leaf.op == ComparisonOp::Eq || leaf.op == ComparisonOp::Contains)
                && leaf.value != Value::Null
        })
        .collect();
    usable
        .iter()
        .find(|leaf| leaf.op == ComparisonOp::Eq)
        .copied()
        .or_else(|| usable.first().copied())
}

struct ExactMatch {
    keys: Vec<String>,
    width: usize,
}

/// Candidates in the order their records first entered the table. The registries hold them in hash
/// sets, which have no order at all, so a query without `orderBy` would answer in a different order
/// every run — where the TS engine, whose registries are insertion-ordered `Map`s and `Set`s,
/// answers in arrival order. Arrival, not last write: overwriting a record does not move it to the
/// back of the page, because `Map.set` on a key already present keeps that key where it was.
fn in_write_order(registry: &TableRegistry, mut keys: Vec<String>) -> Vec<String> {
    keys.sort_by(|left, right| {
        let earlier = registry.arrival_of(left).unwrap_or(0);
        let later = registry.arrival_of(right).unwrap_or(0);
        earlier.cmp(&later).then_with(|| left.cmp(right))
    });
    keys
}

/// Whether the query repeats every `where` clause scoping `index` — an index covering only part of
/// the table may only answer a query that provably falls inside that part.
fn repeats_predicate(registry: &TableRegistry, index: &RegistryIndex, leaves: &[&Predicate]) -> bool {
    let Some(clauses) = registry.where_of(index) else {
        return true;
    };
    clauses.iter().all(|clause| {
        leaves
            .iter()
            .any(|leaf| leaf.field == clause.field && leaf.op == clause.op && same_value(&leaf.value, &clause.value))
    })
}

// NOTE: 0017 narrowed — an empty bucket proves absence only where nothing kept a record out of it.
fn exact_match(registry: &TableRegistry, index: &RegistryIndex, leaves: &[&Predicate]) -> Option<ExactMatch> {
    if index.kind != RegistryIndexKind::Exact && index.kind != RegistryIndexKind::Sorted {
        return None;
    }
    let mut values: Vec<Value> = Vec::new();
    let mut proves = registry.where_of(index).is_none();
    for field in &index.fields {
        match equality_on(leaves, field) {
            Some(predicate) => {
                values.push(predicate.value.clone());
                if !is_primitive_equality(predicate) {
                    proves = false;
                }
            }
            None => break,
        }
    }
    if values.is_empty() {
        return None;
    }
    let width = values.len();
    match registry.candidates_of(index, &values) {
        Some(bucket) => Some(ExactMatch {
            keys: in_write_order(registry, bucket.iter().cloned().collect()),
            width,
        }),
        None => proves.then(|| ExactMatch {
            keys: Vec::new(),
            width,
        }),
    }
}

#[derive(Default)]
struct RawBounds {
    gt: Option<Value>,
    gte: Option<Value>,
    lt: Option<Value>,
    lte: Option<Value>,
}

fn bounds_of(leaves: &[&Predicate], field: &str) -> Option<RawBounds> {
    let mut bounds = RawBounds::default();
    let mut found = false;
    for predicate in leaves {
        if predicate.field != field {
            continue;
        }
        if !matches!(predicate.value, Value::Number(_) | Value::String(_)) {
            continue;
        }
        match predicate.op {
            ComparisonOp::Gt => {
                bounds.gt = Some(predicate.value.clone());
                found = true;
            }
            ComparisonOp::Gte => {
                bounds.gte = Some(predicate.value.clone());
                found = true;
            }
            ComparisonOp::Lt => {
                bounds.lt = Some(predicate.value.clone());
                found = true;
            }
            ComparisonOp::Lte => {
                bounds.lte = Some(predicate.value.clone());
                found = true;
            }
            _ => {}
        }
    }
    found.then_some(bounds)
}

fn with_prefix(prefix: &[SortValue], bounds: &RawBounds) -> RangeBounds {
    let vector_of = |edge: &Option<Value>| -> Option<Vec<SortValue>> {
        let value = edge.as_ref()?;
        let sortable = SortValue::from_json(value)?;
        let mut vector = prefix.to_vec();
        vector.push(sortable);
        Some(vector)
    };
    RangeBounds {
        gt: vector_of(&bounds.gt),
        gte: vector_of(&bounds.gte),
        lt: vector_of(&bounds.lt),
        lte: vector_of(&bounds.lte),
    }
}

fn sorted_keys(registry: &TableRegistry, index: &RegistryIndex, leaves: &[&Predicate]) -> Option<Vec<String>> {
    if index.kind != RegistryIndexKind::Sorted {
        return None;
    }
    let mut prefix: Vec<SortValue> = Vec::new();
    for field in &index.fields {
        let predicate = leaves
            .iter()
            .find(|leaf| leaf.field == *field && is_primitive_equality(leaf));
        match predicate {
            Some(predicate) => prefix.push(SortValue::from_json(&predicate.value)?),
            None => break,
        }
    }
    let open = index.fields.get(prefix.len())?;
    let bounds = bounds_of(leaves, open)?;
    let extended = with_prefix(&prefix, &bounds);
    registry.range_of(index, &extended)
}

fn key_set_of(registry: &TableRegistry, index: &RegistryIndex, strategy: QueryStrategy, keys: Vec<String>) -> KeySet {
    let named = index.fields.len() > 1 || registry.where_of(index).is_some();
    KeySet {
        strategy,
        field: (!named).then(|| index.fields[0].clone()),
        index: named.then(|| index.name.clone()),
        keys,
        branches: None,
    }
}

fn primary_keys_of(leaves: &[&Predicate], primary_key: &str) -> Option<Vec<String>> {
    for predicate in leaves {
        if predicate.field != primary_key {
            continue;
        }
        match (predicate.op, &predicate.value) {
            (ComparisonOp::Eq, Value::String(s)) => return Some(vec![s.clone()]),
            (ComparisonOp::Eq, Value::Number(n)) => return Some(vec![n.to_string()]),
            (ComparisonOp::In, Value::Array(items)) => {
                return Some(items.iter().map(value_as_key).collect());
            }
            _ => {}
        }
    }
    None
}

fn value_as_key(value: &Value) -> String {
    match value {
        Value::String(s) => s.clone(),
        other => other.to_string(),
    }
}

fn primary_path(registry: &TableRegistry, leaves: &[&Predicate]) -> Option<KeySet> {
    let wanted = primary_keys_of(leaves, &registry.schema.primary_key)?;
    let keys: Vec<String> = wanted
        .into_iter()
        .filter(|key| registry.slot_of(key).is_some())
        .collect();
    Some(KeySet {
        strategy: QueryStrategy::Primary,
        field: Some(registry.schema.primary_key.clone()),
        index: None,
        keys,
        branches: None,
    })
}

fn fulltext_path(scores: &HashMap<String, f64>) -> KeySet {
    // NOTE: scores live in a HashMap, so the candidates are ranked here or their order is random.
    let mut keys: Vec<String> = scores.keys().cloned().collect();
    keys.sort_by(|a, b| rank(scores, a, b));
    KeySet {
        strategy: QueryStrategy::Fulltext,
        field: None,
        index: None,
        keys,
        branches: None,
    }
}

fn rank(scores: &HashMap<String, f64>, a: &str, b: &str) -> std::cmp::Ordering {
    let left = scores.get(a).copied().unwrap_or(0.0);
    let right = scores.get(b).copied().unwrap_or(0.0);
    right
        .partial_cmp(&left)
        .unwrap_or(std::cmp::Ordering::Equal)
        .then_with(|| a.cmp(b))
}

fn indexed_path(
    registry: &TableRegistry,
    nodes: &[PredicateNode],
    scores: Option<&HashMap<String, f64>>,
) -> Option<KeySet> {
    let leaves = leaves_of(nodes);
    let mut best: Option<KeySet> = None;
    let mut widest = 0usize;

    let usable: Vec<&RegistryIndex> = registry
        .indexes()
        .iter()
        .filter(|index| repeats_predicate(registry, index, &leaves))
        .collect();

    for index in &usable {
        let Some(matched) = exact_match(registry, index, &leaves) else {
            continue;
        };
        let tied = best.as_ref().is_some_and(|b| matched.keys.len() == b.keys.len());
        let cheaper_than_best = best.as_ref().is_none_or(|b| matched.keys.len() < b.keys.len());
        if tied {
            if matched.width > widest {
                widest = matched.width;
                best = Some(key_set_of(registry, index, QueryStrategy::Exact, matched.keys));
            }
        } else if cheaper_than_best {
            widest = matched.width;
            best = Some(key_set_of(registry, index, QueryStrategy::Exact, matched.keys));
        }
    }

    if let Some(scores) = scores {
        best = cheaper(best, Some(fulltext_path(scores)));
    }

    for index in &usable {
        let Some(keys) = sorted_keys(registry, index, &leaves) else {
            continue;
        };
        best = cheaper(best, Some(key_set_of(registry, index, QueryStrategy::Sorted, keys)));
    }

    // NOTE: nested groups never see `scores` — a top-level search only narrows the top level.
    for node in nodes {
        if matches!(node, PredicateNode::Group { .. }) {
            best = cheaper(best, plan(registry, node, None));
        }
    }

    best
}

fn and_path(
    registry: &TableRegistry,
    nodes: &[PredicateNode],
    scores: Option<&HashMap<String, f64>>,
) -> Option<KeySet> {
    let leaves = leaves_of(nodes);
    cheaper(primary_path(registry, &leaves), indexed_path(registry, nodes, scores))
}

fn branches_of(path: &KeySet) -> Vec<PlanBranch> {
    path.branches.clone().unwrap_or_else(|| {
        vec![PlanBranch {
            strategy: path.strategy,
            field: path.field.clone(),
            candidates: path.keys.len(),
        }]
    })
}

// NOTE: a missing branch (no index, no primary hit) degrades the whole bracket, not just one arm.
fn or_path(registry: &TableRegistry, nodes: &[PredicateNode]) -> Option<KeySet> {
    if nodes.is_empty() {
        return None;
    }
    let mut seen = HashSet::new();
    let mut keys = Vec::new();
    let mut branches = Vec::new();
    for node in nodes {
        let path = plan(registry, node, None)?;
        for key in &path.keys {
            if seen.insert(key.clone()) {
                keys.push(key.clone());
            }
        }
        branches.extend(branches_of(&path));
    }
    Some(KeySet {
        strategy: QueryStrategy::Union,
        field: None,
        index: None,
        keys,
        branches: Some(branches),
    })
}

fn plan(registry: &TableRegistry, node: &PredicateNode, scores: Option<&HashMap<String, f64>>) -> Option<KeySet> {
    match node {
        PredicateNode::Leaf(_) => and_path(registry, std::slice::from_ref(node), scores),
        PredicateNode::Group {
            combine: super::predicates::Combine::And,
            nodes,
        } => and_path(registry, nodes, scores),
        PredicateNode::Group {
            combine: super::predicates::Combine::Or,
            nodes,
        } => or_path(registry, nodes),
    }
}

fn committed_path(registry: &TableRegistry, spec: &QuerySpec, scores: Option<&HashMap<String, f64>>) -> KeySet {
    plan(registry, &spec.root, scores).unwrap_or_else(|| KeySet {
        strategy: QueryStrategy::Scan,
        field: None,
        index: None,
        keys: in_write_order(registry, registry.keys().cloned().collect()),
        branches: None,
    })
}

pub fn plan_access(registry: &TableRegistry, spec: &QuerySpec) -> AccessPath {
    let scores = spec
        .search
        .as_ref()
        .filter(|_| registry.has_fulltext())
        .map(|search| registry.search(&search.text, search.mode));
    committed_path(registry, spec, scores.as_ref()).into_path(scores)
}

/// The top-level conjunctive leaves of `spec` — the only predicates an overlay may narrow on, since
/// an `or` arm proves nothing about the answer as a whole.
fn top_level_leaves(spec: &QuerySpec) -> Vec<&Predicate> {
    match &spec.root {
        PredicateNode::Leaf(leaf) => vec![leaf],
        PredicateNode::Group {
            combine: super::predicates::Combine::And,
            nodes,
        } => nodes
            .iter()
            .filter_map(|node| match node {
                PredicateNode::Leaf(leaf) => Some(leaf),
                PredicateNode::Group { .. } => None,
            })
            .collect(),
        PredicateNode::Group { .. } => Vec::new(),
    }
}

/// The narrowest set of staged keys the overlay's own exact index can prove for `spec`, or `None`
/// when no top-level equality names a field the overlay indexes.
fn indexed_staged(overlay: &dyn QueryOverlay, spec: &QuerySpec) -> Option<Vec<String>> {
    let mut best: Option<Vec<String>> = None;
    for leaf in top_level_leaves(spec) {
        if !is_primitive_equality(leaf) {
            continue;
        }
        let OverlayCandidates::Narrowed(keys) = overlay.candidates_for(&leaf.field, &value_key(&leaf.value)) else {
            continue;
        };
        if best.as_ref().is_none_or(|found| keys.len() < found.len()) {
            best = Some(keys);
        }
    }
    best
}

fn staged_candidates(
    path: &KeySet,
    overlay: &dyn QueryOverlay,
    spec: &QuerySpec,
    scores: Option<&HashMap<String, f64>>,
) -> (Vec<String>, OverlayStrategy) {
    // NOTE: only the fulltext path filters candidates by the query text, so only it may narrow here.
    if path.strategy == QueryStrategy::Fulltext {
        if let Some(scores) = scores {
            let mut keys: Vec<String> = overlay
                .records()
                .keys()
                .filter(|key| scores.contains_key(*key))
                .cloned()
                .collect();
            keys.sort_by(|a, b| rank(scores, a, b));
            return (keys, OverlayStrategy::Fulltext);
        }
    }
    match indexed_staged(overlay, spec) {
        Some(keys) => (keys, OverlayStrategy::Exact),
        None => (overlay.records().keys().cloned().collect(), OverlayStrategy::Scan),
    }
}

/// Plans `spec` as the unit of work will leave the table: the committed candidates the overlay has
/// not taken over, plus the staged records that can still match, and a [`PlanOverlay`] saying how
/// the two halves met. `scores`, when present, were computed over the overlaid corpus, so a staged
/// record joins the ranking rather than merely the answer.
pub fn plan_access_overlaid(
    registry: &TableRegistry,
    spec: &QuerySpec,
    scores: Option<HashMap<String, f64>>,
    overlay: &dyn QueryOverlay,
) -> AccessPath {
    let path = committed_path(registry, spec, scores.as_ref());
    let kept: Vec<String> = path
        .keys
        .iter()
        .filter(|key| !overlay.records().contains_key(*key) && !overlay.deleted().contains(*key))
        .cloned()
        .collect();
    let suppressed = path.keys.len() - kept.len();
    let (staged, strategy) = staged_candidates(&path, overlay, spec, scores.as_ref());
    let summary = PlanOverlay {
        staged: staged.len(),
        suppressed,
        rescored: scores.is_some(),
        strategy,
        size: overlay.records().len(),
    };
    let mut keys = kept;
    keys.extend(staged);
    AccessPath {
        keys,
        overlay: Some(summary),
        ..path.into_path(scores)
    }
}

/// Orders the matched `entries` — each a key paired with the value of the query's sort field, or
/// `Value::Null` when the query names none — and then cuts out the page `offset`/`limit` ask for.
///
/// The order is the one the TS `executeQuery` fixes: an explicit `orderBy` wins; failing that a
/// fulltext query ranks by score, ties broken on the key; failing that the access path's own order
/// stands. Slicing happens last, over the whole ordered result, never before it.
pub fn order_and_page(
    entries: Vec<(String, Value)>,
    spec: &QuerySpec,
    scores: Option<&HashMap<String, f64>>,
) -> Vec<String> {
    let mut entries = entries;
    match (&spec.order_by, scores) {
        (Some(order), _) => {
            let descending = order.direction == SortDirection::Desc;
            entries.sort_by(|(_, left), (_, right)| {
                let ordering = super::predicates::compare_values(left, right);
                if descending {
                    ordering.reverse()
                } else {
                    ordering
                }
            });
        }
        (None, Some(scores)) => entries.sort_by(|(left, _), (right, _)| rank(scores, left, right)),
        (None, None) => {}
    }
    page_of(entries.into_iter().map(|(key, _)| key).collect(), spec)
}

/// The `offset`/`limit` window of `keys`. An offset past the end is an empty page, not an error,
/// and a `limit` of zero selects nothing — both mirror the TS `Array.slice` this replaces.
pub fn page_of(keys: Vec<String>, spec: &QuerySpec) -> Vec<String> {
    if spec.offset.is_none() && spec.limit.is_none() {
        return keys;
    }
    let from = spec.offset.unwrap_or(0).min(keys.len());
    let to = match spec.limit {
        Some(limit) => from.saturating_add(limit).min(keys.len()),
        None => keys.len(),
    };
    keys[from..to].to_vec()
}

/// What a `RecordSource` can fail with. Its own type rather than `io::Error` so a failed decrypt
/// keeps the whole `HaysonCipherError` — key id included — on the way out of the read path;
/// `TableError` cannot be used here because it lives above `query/` in the layering.
#[derive(Debug)]
pub enum RecordSourceError {
    /// Boxed for the same reason as `TableError::Cipher` — only a `cypher` field ever builds one.
    Cipher(Box<HaysonCipherError>),
    Io(io::Error),
}

impl fmt::Display for RecordSourceError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            RecordSourceError::Cipher(error) => error.fmt(f),
            RecordSourceError::Io(error) => error.fmt(f),
        }
    }
}

impl std::error::Error for RecordSourceError {}

impl From<io::Error> for RecordSourceError {
    fn from(value: io::Error) -> Self {
        RecordSourceError::Io(value)
    }
}

impl From<HaysonCipherError> for RecordSourceError {
    fn from(value: HaysonCipherError) -> Self {
        RecordSourceError::Cipher(Box::new(value))
    }
}

/// Extension point for the future `txn/` overlay: today records come straight from the committed
/// log, but an overlay-aware source can be swapped in here once the unit-of-work layer exists.
pub trait RecordSource {
    fn load(&mut self, key: &str) -> Result<Option<HaysonRecord>, RecordSourceError>;
}

/// Decrypts `cypher` fields the moment a record is parsed off the log — see
/// [`LogRecordSource::with_open`].
pub type OpenRecordFn<'a> = dyn Fn(&str, HaysonRecord) -> Result<HaysonRecord, RecordSourceError> + 'a;

#[allow(dead_code)]
pub struct LogRecordSource<'a> {
    registry: &'a TableRegistry,
    log: &'a LogFile,
    /// Applied to a record right after it is parsed off the log — the `cypher`-field decryption
    /// hook a table with a keyring attaches via [`LogRecordSource::with_open`]. `None` for a table
    /// that declares no `cypher` field, which needs no key at all to read.
    open: Option<&'a OpenRecordFn<'a>>,
}

#[allow(dead_code)]
impl<'a> LogRecordSource<'a> {
    pub fn new(registry: &'a TableRegistry, log: &'a LogFile) -> Self {
        LogRecordSource {
            registry,
            log,
            open: None,
        }
    }

    /// Same as [`LogRecordSource::new`], but every record `load` returns is first passed through
    /// `open` (keyed by the record's own key) — used to decrypt `cypher` fields before a record
    /// leaves the log-read boundary, so `get`/`query`/`patch` all hydrate through the same hook.
    pub fn with_open(registry: &'a TableRegistry, log: &'a LogFile, open: &'a OpenRecordFn<'a>) -> Self {
        LogRecordSource {
            registry,
            log,
            open: Some(open),
        }
    }
}

impl RecordSource for LogRecordSource<'_> {
    fn load(&mut self, key: &str) -> Result<Option<HaysonRecord>, RecordSourceError> {
        let Some(entry) = self.registry.slot_of(key) else {
            return Ok(None);
        };
        let slot = Slot {
            offset: entry.offset,
            length: entry.length,
        };
        let line = self.log.read_slice(&slot)?;
        let path = self.log.path().to_string_lossy().into_owned();
        let parsed =
            parse_entry(&line, &path, entry.offset).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
        match (parsed.data, self.open) {
            (Some(record), Some(open)) => Ok(Some(open(key, record)?)),
            (data, _) => Ok(data),
        }
    }
}

/// [`LogRecordSource`] for a table held as a [`LogSet`]: the registry says which partition a key
/// lives in, and the record is read out of that partition's own file. It merges nothing itself —
/// the planner hands it one key at a time, and the registry it plans against already spans every
/// partition of the table.
pub struct LogSetRecordSource<'a> {
    registry: &'a TableRegistry,
    logs: &'a LogSet,
    open: Option<&'a OpenRecordFn<'a>>,
}

impl<'a> LogSetRecordSource<'a> {
    pub fn new(registry: &'a TableRegistry, logs: &'a LogSet) -> Self {
        LogSetRecordSource {
            registry,
            logs,
            open: None,
        }
    }

    /// Same as [`LogSetRecordSource::new`] with the `cypher`-field decryption hook of
    /// [`LogRecordSource::with_open`].
    pub fn with_open(registry: &'a TableRegistry, logs: &'a LogSet, open: &'a OpenRecordFn<'a>) -> Self {
        LogSetRecordSource {
            registry,
            logs,
            open: Some(open),
        }
    }
}

impl RecordSource for LogSetRecordSource<'_> {
    fn load(&mut self, key: &str) -> Result<Option<HaysonRecord>, RecordSourceError> {
        let Some(entry) = self.registry.slot_of(key) else {
            return Ok(None);
        };
        let part = self.registry.part_of(key).unwrap_or(UNPARTITIONED);
        let slot = Slot {
            offset: entry.offset,
            length: entry.length,
        };
        let line = self.logs.read_slice(part, &slot)?;
        let path = self.logs.path_of(part).to_string_lossy().into_owned();
        let parsed =
            parse_entry(&line, &path, entry.offset).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
        match (parsed.data, self.open) {
            (Some(record), Some(open)) => Ok(Some(open(key, record)?)),
            (data, _) => Ok(data),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QueryResult {
    /// The page the query asked for: matched, ordered, then cut down by `offset`/`limit`. Empty for
    /// a `count_only` spec, which asked for no page at all.
    pub keys: Vec<String>,
    /// How many records matched before `offset`/`limit` cut the page out of them.
    pub matched: usize,
    pub plan: AccessPathSummary,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AccessPathSummary {
    pub strategy: QueryStrategy,
    pub field: Option<String>,
    pub index: Option<String>,
    pub candidates: usize,
    /// The answer came out in relevance order: a `search` scored the candidates and no explicit
    /// `order_by` overrode that order.
    pub ranked: bool,
    /// Every field the query touches lives in the registry's projection, so the predicate was
    /// tested against cached values and no record was read off the log.
    pub filtered_on_projection: bool,
    pub reads: usize,
    pub branches: Option<Vec<PlanBranch>>,
    /// Set only for a query run through a unit of work — see [`PlanOverlay`].
    pub overlay: Option<PlanOverlay>,
}

/// Every field a query reads to answer itself: the fields of every predicate leaf, nested groups
/// included, plus the sort field. Mirrors the `fieldsInPlay` set of the TS `executeQuery`.
fn fields_in_play(spec: &QuerySpec) -> Vec<&str> {
    let mut fields: Vec<&str> = all_leaves(std::slice::from_ref(&spec.root))
        .iter()
        .map(|leaf| leaf.field.as_str())
        .collect();
    if let Some(order) = &spec.order_by {
        fields.push(order.field.as_str());
    }
    fields
}

/// Whether a `spec.search` still admits `key`. A committed candidate is admitted by the candidate
/// set itself only when the plan took the fulltext path; a staged one is never, since an index on
/// the overlay may have picked it. Answered off the scores, which a unit of work computes over the
/// overlaid corpus, so the staged record is in them exactly when the search matches it.
fn scored_in(access: &AccessPath, spec: &QuerySpec, key: &str) -> bool {
    if spec.search.is_none() {
        return true;
    }
    access.scores.as_ref().is_some_and(|scores| scores.contains_key(key))
}

fn matched_entry(
    record: &HaysonRecord,
    key: &str,
    spec: &QuerySpec,
    sort_field: Option<&str>,
) -> Option<(String, Value)> {
    if !test_node(record, &spec.root) {
        return None;
    }
    let sort_value = sort_field.map_or(Value::Null, |field| record.get(field).cloned().unwrap_or(Value::Null));
    Some((key.to_string(), sort_value))
}

/// Runs `spec` against `registry`, narrowing candidates through `plan_access` and then testing the
/// rest of the predicate tree against records pulled one at a time from `source`. A candidate set
/// of zero never touches `source` — that is the load-bearing behaviour of ADR-dsam/0017.
///
/// A query whose every field is projected never touches `source` either: the predicate is tested
/// against the registry's in-memory projections and `reads` stays at zero, the same shortcut the
/// TS `executeQuery` takes.
pub fn execute_query<S: RecordSource + ?Sized>(
    registry: &TableRegistry,
    source: &mut S,
    spec: &QuerySpec,
) -> Result<QueryResult, RecordSourceError> {
    let access = plan_access(registry, spec);
    execute_access(registry, source, spec, access, None)
}

/// [`execute_query`] over an access path the caller already planned, with the records of `overlay`
/// answered from the overlay itself instead of the log — the `txn/` half of the extension point
/// [`RecordSource`] names. A staged record costs no read and is never tested on a registry
/// projection, because the registry has not been told about it yet.
pub fn execute_access<S: RecordSource + ?Sized>(
    registry: &TableRegistry,
    source: &mut S,
    spec: &QuerySpec,
    access: AccessPath,
    overlay: Option<&dyn QueryOverlay>,
) -> Result<QueryResult, RecordSourceError> {
    let sort_field = spec.order_by.as_ref().map(|order| order.field.as_str());
    let on_projection = fields_in_play(spec)
        .into_iter()
        .all(|field| registry.is_projected(field));
    let mut matched: Vec<(String, Value)> = Vec::new();
    let mut reads = 0usize;
    for key in &access.keys {
        let staged = overlay.and_then(|overlay| overlay.records().get(key));
        let entry = match staged {
            // NOTE: only the fulltext path search-filters its candidates, so a staged key re-checks here.
            Some(_) if !scored_in(&access, spec, key) => None,
            Some(record) => matched_entry(record, key, spec, sort_field),
            None if on_projection => registry
                .projection_of(key)
                .and_then(|projection| matched_entry(projection, key, spec, sort_field)),
            None => match source.load(key)? {
                Some(record) => {
                    reads += 1;
                    matched_entry(&record, key, spec, sort_field)
                }
                None => None,
            },
        };
        if let Some(entry) = entry {
            matched.push(entry);
        }
    }
    let total = matched.len();
    let keys = if spec.count_only {
        Vec::new()
    } else {
        order_and_page(matched, spec, access.scores.as_ref())
    };
    Ok(QueryResult {
        keys,
        matched: total,
        plan: AccessPathSummary {
            strategy: access.strategy,
            field: access.field,
            index: access.index,
            candidates: access.keys.len(),
            ranked: access.scores.is_some() && spec.order_by.is_none(),
            filtered_on_projection: on_projection,
            reads,
            branches: access.branches,
            overlay: access.overlay,
        },
    })
}

#[cfg(test)]
mod tests {
    use serde_json::json;

    use super::*;
    use crate::query::builder::QueryBuilder;
    use crate::schema::parse_schema;
    use crate::storage::record::{serialize_entry, LogMeta, LogOp};

    struct CountingSource<'a> {
        inner: LogRecordSource<'a>,
        reads: usize,
    }

    impl RecordSource for CountingSource<'_> {
        fn load(&mut self, key: &str) -> Result<Option<HaysonRecord>, RecordSourceError> {
            self.reads += 1;
            self.inner.load(key)
        }
    }

    fn users_schema() -> crate::schema::TableSchema {
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "name": { "type": "string" },
                "age": { "type": "int", "index": "sorted" },
                "role": { "type": "string", "index": "exact" },
                "tags": { "type": "string[]", "index": "exact" },
            },
        }))
        .unwrap()
    }

    fn record(pairs: &[(&str, Value)]) -> HaysonRecord {
        pairs.iter().map(|(k, v)| (k.to_string(), v.clone())).collect()
    }

    fn ania() -> HaysonRecord {
        record(&[
            ("id", json!("user_1")),
            ("name", json!("Ania")),
            ("age", json!(31)),
            ("role", json!("admin")),
            ("tags", json!(["backend", "node"])),
        ])
    }

    fn piotr() -> HaysonRecord {
        record(&[
            ("id", json!("user_2")),
            ("name", json!("Piotr")),
            ("age", json!(24)),
            ("role", json!("user")),
            ("tags", json!(["frontend"])),
        ])
    }

    fn maria() -> HaysonRecord {
        record(&[
            ("id", json!("user_3")),
            ("name", json!("Maria")),
            ("age", json!(41)),
            ("role", json!("user")),
            ("tags", json!(["backend", "go"])),
        ])
    }

    struct Fixture {
        _dir: tempfile::TempDir,
        registry: TableRegistry,
        log: LogFile,
    }

    fn fixture_with_schema(schema: crate::schema::TableSchema, records: &[HaysonRecord]) -> Fixture {
        fixture_of(TableRegistry::new(schema), records)
    }

    fn fixture_of(registry: TableRegistry, records: &[HaysonRecord]) -> Fixture {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("users.jsonl");
        let mut log = LogFile::open(&path, false).unwrap();
        let mut registry = registry;

        let mut seq = 0u64;
        for entry in records {
            seq += 1;
            let key = entry.get("id").and_then(Value::as_str).unwrap().to_string();
            let line = serialize_entry(
                &LogMeta {
                    key: key.clone(),
                    op: LogOp::Put,
                    seq,
                    ts: seq as i64,
                },
                Some(entry),
            );
            let slots = log.append(&[line]).unwrap();
            registry.apply_put(&key, entry, slots[0], seq);
        }
        Fixture {
            _dir: dir,
            registry,
            log,
        }
    }

    fn fixture(records: &[HaysonRecord]) -> Fixture {
        fixture_with_schema(users_schema(), records)
    }

    fn fulltext_users_schema() -> crate::schema::TableSchema {
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "name": { "type": "string" },
                "bio": { "type": "string", "fulltext": true },
                "age": { "type": "int", "index": "sorted" },
                "role": { "type": "string", "index": "exact" },
            },
        }))
        .unwrap()
    }

    fn ania_bio() -> HaysonRecord {
        record(&[
            ("id", json!("user_1")),
            ("name", json!("Ania")),
            ("bio", json!("Backend developer, Łódź")),
            ("age", json!(31)),
            ("role", json!("admin")),
        ])
    }

    fn piotr_bio() -> HaysonRecord {
        record(&[
            ("id", json!("user_2")),
            ("name", json!("Piotr")),
            ("bio", json!("Frontend developer from Warsaw")),
            ("age", json!(24)),
            ("role", json!("user")),
        ])
    }

    fn maria_bio() -> HaysonRecord {
        record(&[
            ("id", json!("user_3")),
            ("name", json!("Maria")),
            ("bio", json!("Backend developer, Kraków")),
            ("age", json!(41)),
            ("role", json!("user")),
        ])
    }

    fn fulltext_fixture() -> Fixture {
        fixture_with_schema(fulltext_users_schema(), &[ania_bio(), piotr_bio(), maria_bio()])
    }

    fn overwrite(fixture: &mut Fixture, entry: &HaysonRecord, seq: u64) {
        let key = entry.get("id").and_then(Value::as_str).unwrap().to_string();
        let line = serialize_entry(
            &LogMeta {
                key: key.clone(),
                op: LogOp::Put,
                seq,
                ts: seq as i64,
            },
            Some(entry),
        );
        let slots = fixture.log.append(&[line]).unwrap();
        fixture.registry.apply_put(&key, entry, slots[0], seq);
    }

    fn run(fixture: &Fixture, spec: &QuerySpec) -> (QueryResult, usize) {
        let mut source = CountingSource {
            inner: LogRecordSource::new(&fixture.registry, &fixture.log),
            reads: 0,
        };
        let result = execute_query(&fixture.registry, &mut source, spec).unwrap();
        (result, source.reads)
    }

    #[test]
    fn uses_the_exact_index_for_a_single_predicate() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("role", ComparisonOp::Eq, json!("user"))
            .spec();

        let (result, reads) = run(&fixture, &spec);
        assert_eq!(result.plan.strategy, QueryStrategy::Exact);
        assert_eq!(result.plan.field, Some("role".to_string()));
        assert_eq!(result.plan.candidates, 2);
        assert_eq!(reads, 0, "`role` is projected, so the predicate never leaves memory");
        let mut ids = result.keys.clone();
        ids.sort();
        assert_eq!(ids, vec!["user_2", "user_3"]);
    }

    #[test]
    fn narrows_through_the_index_then_filters_the_rest_in_memory() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("role", ComparisonOp::Eq, json!("user"))
            .r#where("age", ComparisonOp::Gt, json!(30))
            .spec();

        let (result, reads) = run(&fixture, &spec);
        assert_eq!(result.plan.strategy, QueryStrategy::Exact);
        assert_eq!(result.plan.candidates, 2);
        assert_eq!(
            reads, 0,
            "`role` and `age` are both projected — the second filter runs on projections too"
        );
        assert_eq!(result.keys, vec!["user_3".to_string()]);
    }

    #[test]
    fn picks_the_more_selective_index_when_more_than_one_applies() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("role", ComparisonOp::Eq, json!("admin"))
            .r#where("age", ComparisonOp::Gt, json!(10))
            .spec();

        let (result, _) = run(&fixture, &spec);
        assert_eq!(result.plan.strategy, QueryStrategy::Exact);
        assert_eq!(result.plan.field, Some("role".to_string()));
        assert_eq!(result.plan.candidates, 1);
    }

    #[test]
    fn unions_the_branches_of_an_or_bracket() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .where_any(|group| {
                group.r#where("role", ComparisonOp::Eq, json!("admin")).r#where(
                    "tags",
                    ComparisonOp::Contains,
                    json!("go"),
                );
            })
            .spec();

        let (result, _) = run(&fixture, &spec);
        assert_eq!(result.plan.strategy, QueryStrategy::Union);
        assert_eq!(result.plan.candidates, 2);
        assert_eq!(
            result.plan.branches,
            Some(vec![
                PlanBranch {
                    strategy: QueryStrategy::Exact,
                    field: Some("role".to_string()),
                    candidates: 1
                },
                PlanBranch {
                    strategy: QueryStrategy::Exact,
                    field: Some("tags".to_string()),
                    candidates: 1
                },
            ])
        );
        let mut ids = result.keys.clone();
        ids.sort();
        assert_eq!(ids, vec!["user_1", "user_3"]);
    }

    #[test]
    fn nests_an_and_bracket_inside_an_or_bracket() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .where_any(|group| {
                group
                    .r#where("role", ComparisonOp::Eq, json!("admin"))
                    .where_all(|inner| {
                        inner.r#where("role", ComparisonOp::Eq, json!("user")).r#where(
                            "age",
                            ComparisonOp::Gt,
                            json!(35),
                        );
                    });
            })
            .spec();

        let (result, _) = run(&fixture, &spec);
        assert_eq!(result.plan.strategy, QueryStrategy::Union);
        assert_eq!(
            result.plan.branches,
            Some(vec![
                PlanBranch {
                    strategy: QueryStrategy::Exact,
                    field: Some("role".to_string()),
                    candidates: 1
                },
                PlanBranch {
                    strategy: QueryStrategy::Sorted,
                    field: Some("age".to_string()),
                    candidates: 1
                },
            ])
        );
        let mut ids = result.keys.clone();
        ids.sort();
        assert_eq!(ids, vec!["user_1", "user_3"]);
    }

    #[test]
    fn degrades_the_whole_bracket_to_a_scan_when_one_branch_has_no_index() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .where_any(|group| {
                group.r#where("role", ComparisonOp::Eq, json!("admin")).r#where(
                    "name",
                    ComparisonOp::StartsWith,
                    json!("Piotr"),
                );
            })
            .spec();

        let (result, _) = run(&fixture, &spec);
        assert_eq!(result.plan.strategy, QueryStrategy::Scan);
        assert_eq!(result.plan.candidates, 3);
        assert_eq!(result.plan.branches, None);
        let mut ids = result.keys.clone();
        ids.sort();
        assert_eq!(ids, vec!["user_1", "user_2"]);
    }

    /// ADR-dsam/0017 — an equality miss on an indexed field is a proven-empty result, not a
    /// license to fall back to a full scan. Verified two ways: the plan's candidate count is
    /// zero, AND the record source (backed by the real log file) is never asked to load anything.
    #[test]
    fn an_empty_index_bucket_never_falls_back_to_a_scan() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("role", ComparisonOp::Eq, json!("ghost"))
            .spec();

        let (result, reads) = run(&fixture, &spec);
        assert_eq!(result.plan.strategy, QueryStrategy::Exact);
        assert_eq!(result.plan.field, Some("role".to_string()));
        assert_eq!(result.plan.candidates, 0);
        assert_eq!(
            reads, 0,
            "the log must not be read when the index already proved an empty result"
        );
        assert!(result.keys.is_empty());
    }

    #[test]
    fn falls_back_to_a_scan_when_no_registry_can_help() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("name", ComparisonOp::StartsWith, json!("A"))
            .spec();

        let (result, reads) = run(&fixture, &spec);
        assert_eq!(result.plan.strategy, QueryStrategy::Scan);
        assert_eq!(result.plan.candidates, 3);
        assert_eq!(reads, 3);
        assert_eq!(result.keys, vec!["user_1".to_string()]);
    }

    #[test]
    fn reaches_the_primary_registry_from_a_top_level_equality() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("id", ComparisonOp::Eq, json!("user_2"))
            .spec();

        let (result, _) = run(&fixture, &spec);
        assert_eq!(result.plan.strategy, QueryStrategy::Primary);
        assert_eq!(result.plan.candidates, 1);
        assert_eq!(result.keys, vec!["user_2".to_string()]);
    }

    #[test]
    fn uses_the_fulltext_index_for_a_search_query() {
        let fixture = fulltext_fixture();
        let spec = QueryBuilder::new().search("backend developer", SearchMode::All).spec();

        let (result, _) = run(&fixture, &spec);
        assert_eq!(result.plan.strategy, QueryStrategy::Fulltext);
        assert_eq!(result.plan.field, None);
        assert_eq!(result.plan.candidates, 2);
        let mut ids = result.keys.clone();
        ids.sort();
        assert_eq!(ids, vec!["user_1", "user_3"]);
    }

    // NOTE: search sits outside the predicate tree, so only the winning strategy's own
    // candidates ever get compared to the search term -- pairing it with a non-indexed field
    // keeps fulltext as the sole access path and lets the other leaf narrow in memory.
    #[test]
    fn narrows_a_fulltext_search_with_an_in_memory_predicate() {
        let fixture = fulltext_fixture();
        let spec = QueryBuilder::new()
            .search("backend developer", SearchMode::All)
            .r#where("name", ComparisonOp::StartsWith, json!("A"))
            .spec();

        let (result, _) = run(&fixture, &spec);
        assert_eq!(result.plan.strategy, QueryStrategy::Fulltext);
        assert_eq!(result.plan.candidates, 2);
        assert_eq!(result.keys, vec!["user_1".to_string()]);
    }

    #[test]
    fn matches_a_direct_registry_search_call() {
        let fixture = fulltext_fixture();
        let spec = QueryBuilder::new().search("backend developer", SearchMode::All).spec();

        let (result, _) = run(&fixture, &spec);
        let mut via_planner = result.keys.clone();
        via_planner.sort();

        let mut direct: Vec<String> = fixture
            .registry
            .search("backend developer", SearchMode::All)
            .into_keys()
            .collect();
        direct.sort();

        assert_eq!(via_planner, direct);
    }

    #[test]
    fn orders_the_result_by_a_field_ascending_and_descending() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let ascending = QueryBuilder::new().order_by("age", SortDirection::Asc).spec();
        let descending = QueryBuilder::new().order_by("age", SortDirection::Desc).spec();

        let (up, _) = run(&fixture, &ascending);
        let (down, _) = run(&fixture, &descending);

        assert_eq!(up.keys, vec!["user_2", "user_1", "user_3"], "24, 31, 41");
        assert_eq!(down.keys, vec!["user_3", "user_1", "user_2"], "41, 31, 24");
    }

    /// The order of the three steps is the whole point: were the page cut before the predicate ran,
    /// the two oldest of every record (`user_1`, `user_3`) would answer instead of the two oldest
    /// of those the predicate kept.
    #[test]
    fn a_page_is_cut_out_of_the_ordered_result_only_after_the_predicate_narrowed_it() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("age", ComparisonOp::Gt, json!(24))
            .order_by("age", SortDirection::Asc)
            .limit(1)
            .spec();

        let (result, _) = run(&fixture, &spec);

        assert_eq!(result.keys, vec!["user_1"], "31 is the youngest of those older than 24");
        assert_eq!(result.matched, 2);
    }

    #[test]
    fn an_offset_and_a_limit_together_select_a_middle_window_of_the_order() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .order_by("age", SortDirection::Asc)
            .offset(1)
            .limit(1)
            .spec();

        let (result, _) = run(&fixture, &spec);

        assert_eq!(result.keys, vec!["user_1"]);
        assert_eq!(
            result.matched, 3,
            "the page is one key wide, the match count counts all three"
        );
    }

    #[test]
    fn an_offset_past_the_end_and_a_limit_of_zero_both_answer_an_empty_page() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let past_the_end = QueryBuilder::new()
            .order_by("age", SortDirection::Asc)
            .offset(99)
            .spec();
        let nothing = QueryBuilder::new().order_by("age", SortDirection::Asc).limit(0).spec();

        let (skipped, _) = run(&fixture, &past_the_end);
        let (none, _) = run(&fixture, &nothing);

        assert!(skipped.keys.is_empty());
        assert_eq!(skipped.matched, 3);
        assert!(none.keys.is_empty());
        assert_eq!(none.matched, 3);
    }

    #[test]
    fn a_page_window_over_an_empty_result_stays_empty() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("role", ComparisonOp::Eq, json!("ghost"))
            .order_by("age", SortDirection::Asc)
            .offset(2)
            .limit(5)
            .spec();

        let (result, _) = run(&fixture, &spec);

        assert!(result.keys.is_empty());
        assert_eq!(result.matched, 0);
    }

    // NOTE: only one sort key exists (as in TS), so a tie is left alone — no second key decides it.
    #[test]
    fn a_tie_on_the_sort_field_keeps_every_matched_key_exactly_once() {
        let mut twin = piotr();
        twin.insert("id".to_string(), json!("user_4"));
        twin.insert("age".to_string(), json!(41));
        let fixture = fixture(&[ania(), piotr(), maria(), twin]);
        let spec = QueryBuilder::new().order_by("age", SortDirection::Asc).spec();

        let (result, _) = run(&fixture, &spec);

        assert_eq!(result.keys.len(), 4);
        assert_eq!(
            &result.keys[..2],
            ["user_2".to_string(), "user_1".to_string()],
            "24 then 31"
        );
        let mut tied = result.keys[2..].to_vec();
        tied.sort();
        assert_eq!(
            tied,
            vec!["user_3", "user_4"],
            "both 41s are present, in whichever order"
        );
    }

    #[test]
    fn a_fulltext_search_ranks_by_score_before_the_page_is_cut() {
        let fixture = fulltext_fixture();
        let ranked = QueryBuilder::new().search("backend developer", SearchMode::All).spec();
        let first = QueryBuilder::new()
            .search("backend developer", SearchMode::All)
            .limit(1)
            .spec();

        let (all, _) = run(&fixture, &ranked);
        let (page, _) = run(&fixture, &first);

        assert_eq!(
            page.keys,
            all.keys[..1],
            "the page is the head of the ranking, not an arbitrary hit"
        );
        assert_eq!(page.matched, 2);
    }

    #[test]
    fn an_explicit_order_replaces_the_fulltext_ranking() {
        let fixture = fulltext_fixture();
        let spec = QueryBuilder::new()
            .search("backend developer", SearchMode::All)
            .order_by("age", SortDirection::Desc)
            .limit(1)
            .spec();

        let (result, _) = run(&fixture, &spec);

        assert_eq!(
            result.keys,
            vec!["user_3"],
            "Maria is 41 — the ranking would have led with user_1"
        );
    }

    #[test]
    fn an_aggregate_folds_every_match_and_not_just_the_page() {
        use crate::query::aggregate::execute_aggregate;

        let fixture = fixture(&[ania(), piotr(), maria()]);
        let (spec, aggregate) = QueryBuilder::new().count().limit(1).build_aggregate();

        let mut source = LogRecordSource::new(&fixture.registry, &fixture.log);
        let folded = execute_aggregate(&fixture.registry, &mut source, &spec, &aggregate).unwrap();

        assert_eq!(folded.count, 3);
    }

    fn named_users_schema() -> crate::schema::TableSchema {
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "name": { "type": "string", "index": "exact" },
                "age": { "type": "int", "index": "sorted" },
                "role": { "type": "string", "index": "exact" },
                "tags": { "type": "string[]", "index": "exact" },
            },
        }))
        .unwrap()
    }

    fn ania_kowalska() -> HaysonRecord {
        record(&[
            ("id", json!("user_1")),
            ("name", json!("Ania Kowalska")),
            ("age", json!(31)),
            ("role", json!("admin")),
            ("tags", json!(["backend", "node"])),
        ])
    }

    /// An `exact` index answers "which keys hold exactly this value" and nothing else, so a
    /// substring test against a field carrying one has to leave it alone — routing `contains`
    /// through it looked up "Kowals" as a whole name, missed, and called the miss a proven absence.
    #[test]
    fn a_contains_predicate_on_a_string_field_with_an_exact_index_falls_back_to_a_scan() {
        let fixture = fixture_with_schema(named_users_schema(), &[ania_kowalska(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("name", ComparisonOp::Contains, json!("Kowals"))
            .spec();

        let (result, _) = run(&fixture, &spec);

        assert_eq!(result.plan.strategy, QueryStrategy::Scan);
        assert_eq!(result.keys, vec!["user_1".to_string()]);
    }

    /// The counterpart of the test above: membership in an array field IS what the exact index
    /// holds, one bucket per element, so that `contains` must keep its index.
    #[test]
    fn a_contains_predicate_on_an_array_field_keeps_using_the_exact_index() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("tags", ComparisonOp::Contains, json!("go"))
            .spec();

        let (result, _) = run(&fixture, &spec);

        assert_eq!(result.plan.strategy, QueryStrategy::Exact);
        assert_eq!(result.plan.field, Some("tags".to_string()));
        assert_eq!(result.plan.candidates, 1);
        assert_eq!(result.keys, vec!["user_3".to_string()]);
    }

    /// An array field is indexed per element, so the whole array is a value the index cannot hold —
    /// its silence about one says nothing, and the query has to be answered by a scan.
    #[test]
    fn an_equality_against_an_array_value_falls_back_to_a_scan() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("tags", ComparisonOp::Eq, json!(["backend", "node"]))
            .spec();

        let (result, _) = run(&fixture, &spec);

        assert_eq!(result.plan.strategy, QueryStrategy::Scan);
        assert_eq!(result.keys, vec!["user_1".to_string()]);
    }

    fn admins_only_index() -> Vec<crate::catalog::ExpandedIndex> {
        vec![crate::catalog::ExpandedIndex {
            name: "users_admin_name".to_string(),
            fields: vec!["name".to_string()],
            kind: crate::catalog::CatalogIndexKind::Exact,
            unique: false,
            where_: Some(vec![crate::catalog::CatalogPredicate {
                field: "role".to_string(),
                op: "=".to_string(),
                value: json!("admin"),
            }]),
            expire_after_seconds: None,
        }]
    }

    fn partial_fixture() -> Fixture {
        let schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "name": { "type": "string" },
                "age": { "type": "int" },
                "role": { "type": "string" },
                "tags": { "type": "string[]" },
            },
        }))
        .unwrap();
        let registry = TableRegistry::with_declared_indexes(schema, &admins_only_index());
        fixture_of(registry, &[ania(), piotr(), maria()])
    }

    /// ADR-dsam/0017 narrowed: a `where`-scoped index holds part of the table, so its empty bucket
    /// may mean "the record was never offered to me" rather than "no such record exists".
    #[test]
    fn a_where_scoped_index_never_answers_a_query_that_omits_its_predicate() {
        let fixture = partial_fixture();
        let spec = QueryBuilder::new()
            .r#where("name", ComparisonOp::Eq, json!("Piotr"))
            .spec();

        let (result, _) = run(&fixture, &spec);

        assert_eq!(
            result.plan.strategy,
            QueryStrategy::Scan,
            "Piotr is not an admin, so the index never saw him"
        );
        assert_eq!(result.plan.candidates, 3);
        assert_eq!(result.keys, vec!["user_2".to_string()]);
    }

    #[test]
    fn a_where_scoped_index_answers_a_query_that_repeats_its_predicate() {
        let fixture = partial_fixture();
        let spec = QueryBuilder::new()
            .r#where("name", ComparisonOp::Eq, json!("Ania"))
            .r#where("role", ComparisonOp::Eq, json!("admin"))
            .spec();

        let (result, _) = run(&fixture, &spec);

        assert_eq!(result.plan.strategy, QueryStrategy::Exact);
        assert_eq!(result.plan.index, Some("users_admin_name".to_string()));
        assert_eq!(
            result.plan.field, None,
            "a scoped index is named by the index, not by its one field"
        );
        assert_eq!(result.plan.candidates, 1);
        assert_eq!(result.keys, vec!["user_1".to_string()]);
    }

    /// A query without `orderBy` answers in the order the records were written — the registries
    /// hold their candidates in hash sets, which would otherwise reorder the answer run to run.
    #[test]
    fn candidates_of_an_index_and_of_a_scan_alike_come_out_in_write_order() {
        let fixture = fixture(&[maria(), ania(), piotr()]);
        let indexed = QueryBuilder::new()
            .r#where("role", ComparisonOp::Eq, json!("user"))
            .spec();
        let scanned = QueryBuilder::new()
            .r#where("name", ComparisonOp::StartsWith, json!(""))
            .spec();

        let (by_index, _) = run(&fixture, &indexed);
        let (by_scan, _) = run(&fixture, &scanned);

        assert_eq!(
            by_index.keys,
            vec!["user_3".to_string(), "user_2".to_string()],
            "Maria was written first"
        );
        assert_eq!(
            by_scan.keys,
            vec!["user_3".to_string(), "user_1".to_string(), "user_2".to_string()],
            "the scan follows the log, not the hash of the key"
        );
    }

    /// NOTE-8bmm: overwriting a record must not shuffle it to the back of an unordered page --
    /// the TS engine's `Map`-backed primary registry keeps an existing key exactly where it was.
    #[test]
    fn an_overwritten_record_keeps_its_place_on_an_unordered_page() {
        let mut fixture = fixture(&[ania(), piotr(), maria()]);
        let mut updated = ania();
        updated.insert("age".to_string(), json!(32));
        overwrite(&mut fixture, &updated, 4);
        let scanned = QueryBuilder::new()
            .r#where("name", ComparisonOp::StartsWith, json!(""))
            .spec();

        let (result, _) = run(&fixture, &scanned);

        assert_eq!(
            result.keys,
            vec!["user_1".to_string(), "user_2".to_string(), "user_3".to_string()],
            "the update rewrote Ania in place, so she stays first -- not last behind Maria"
        );
    }

    #[test]
    fn a_record_written_after_an_overwrite_still_lands_at_the_end_of_the_page() {
        let mut fixture = fixture(&[ania(), piotr()]);
        let mut updated = ania();
        updated.insert("age".to_string(), json!(32));
        overwrite(&mut fixture, &updated, 3);
        overwrite(&mut fixture, &maria(), 4);
        let scanned = QueryBuilder::new()
            .r#where("name", ComparisonOp::StartsWith, json!(""))
            .spec();

        let (result, _) = run(&fixture, &scanned);

        assert_eq!(
            result.keys,
            vec!["user_1".to_string(), "user_2".to_string(), "user_3".to_string()],
            "Maria is genuinely new, so she goes last however recently Ania was rewritten"
        );
    }

    #[test]
    fn an_empty_fulltext_result_never_falls_back_to_a_scan() {
        let fixture = fulltext_fixture();
        let spec = QueryBuilder::new().search("zzz_nowhere", SearchMode::All).spec();

        let (result, reads) = run(&fixture, &spec);
        assert_eq!(result.plan.strategy, QueryStrategy::Fulltext);
        assert_eq!(result.plan.candidates, 0);
        assert_eq!(
            reads, 0,
            "the log must not be read when the search already proved an empty result"
        );
        assert!(result.keys.is_empty());
    }

    #[test]
    fn a_search_left_to_its_own_order_reports_a_ranked_plan() {
        let fixture = fulltext_fixture();
        let spec = QueryBuilder::new().search("backend developer", SearchMode::All).spec();

        let (result, _) = run(&fixture, &spec);

        assert!(result.plan.ranked, "the order came out of the relevance scores");
    }

    #[test]
    fn an_explicit_order_beside_a_search_reports_an_unranked_plan() {
        let fixture = fulltext_fixture();
        let spec = QueryBuilder::new()
            .search("backend developer", SearchMode::All)
            .order_by("age", SortDirection::Desc)
            .spec();

        let (result, _) = run(&fixture, &spec);

        assert!(
            !result.plan.ranked,
            "the scores were computed but the explicit order overrode them"
        );
    }

    #[test]
    fn a_query_without_a_search_reports_an_unranked_plan() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("role", ComparisonOp::Eq, json!("user"))
            .spec();

        let (result, _) = run(&fixture, &spec);

        assert!(!result.plan.ranked, "no search ran, so there is nothing to rank by");
    }

    #[test]
    fn a_query_whose_every_field_is_projected_answers_without_reading_a_record() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("role", ComparisonOp::Eq, json!("user"))
            .r#where("age", ComparisonOp::Gt, json!(30))
            .spec();

        let (result, reads) = run(&fixture, &spec);

        assert!(result.plan.filtered_on_projection);
        assert_eq!(result.plan.reads, 0);
        assert_eq!(reads, 0);
        assert_eq!(result.keys, vec!["user_3".to_string()]);
    }

    #[test]
    fn a_query_touching_a_field_outside_the_projection_reads_every_candidate() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("name", ComparisonOp::Eq, json!("Maria"))
            .spec();

        let (result, reads) = run(&fixture, &spec);

        assert!(
            !result.plan.filtered_on_projection,
            "`name` carries no index, so it is not projected"
        );
        assert_eq!(result.plan.reads, 3);
        assert_eq!(reads, 3);
        assert_eq!(result.keys, vec!["user_3".to_string()]);
    }

    #[test]
    fn a_sort_on_a_field_outside_the_projection_takes_the_query_off_the_projections() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("role", ComparisonOp::Eq, json!("user"))
            .order_by("name", SortDirection::Asc)
            .spec();

        let (result, _) = run(&fixture, &spec);

        assert!(
            !result.plan.filtered_on_projection,
            "the sort field counts as a field in play"
        );
        assert_eq!(result.plan.reads, 2);
        assert_eq!(
            result.keys,
            vec!["user_3".to_string(), "user_2".to_string()],
            "Maria before Piotr"
        );
    }

    #[test]
    fn a_count_only_query_answers_with_the_match_count_and_no_page() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let paged = QueryBuilder::new()
            .r#where("role", ComparisonOp::Eq, json!("user"))
            .spec();
        let counting = paged.clone().counting(true);

        let (page, _) = run(&fixture, &paged);
        let (count, _) = run(&fixture, &counting);

        assert_eq!(
            page.keys.len(),
            2,
            "the same query hands back a page when it is not counting"
        );
        assert_eq!(count.matched, page.matched);
        assert!(
            count.keys.is_empty(),
            "a count-only query names no page for a caller to load"
        );
        assert_eq!(
            count.plan, page.plan,
            "counting changes what is returned, not how it was planned"
        );
    }

    #[test]
    fn a_count_only_query_counts_every_match_even_where_a_page_would_be_cut() {
        let fixture = fixture(&[ania(), piotr(), maria()]);
        let spec = QueryBuilder::new()
            .r#where("role", ComparisonOp::Eq, json!("user"))
            .spec()
            .with_limit(1);

        let (count, _) = run(&fixture, &spec.clone().counting(true));

        assert_eq!(
            count.matched, 2,
            "`matched` counts before `limit` cuts, counting or not"
        );
        assert!(count.keys.is_empty());
    }

    #[test]
    fn an_aggregate_keeps_its_keys_when_the_spec_it_unpages_was_counting() {
        let spec = QueryBuilder::new()
            .r#where("role", ComparisonOp::Eq, json!("user"))
            .spec()
            .counting(true);

        // NOTE: an aggregate folds record by record over `keys`, so it can never run count-only.
        assert!(!spec.unpaged().count_only);
    }
}