lucisearch 0.8.0

//! `Index` — unified facade for creating, writing, and searching a Luci index.
//!
//! This is the primary entry point for using Luci. It wires together
//! storage, schema, analyzers, writer, and searcher into a single API.
//!
//! ```ignore
//! use luci::search::expression::SearchExpression;
//!
//! let mut index = Index::create("/tmp/my_index", schema)?;
//! index.add(json!({"title": "hello world", "status": "active"}))?;
//! let expr = SearchExpression::from_json(json!({"match": {"title": "hello"}}), 10)?;
//! let results = index.search(&expr)?;
//! ```

use std::path::Path;

use crate::analysis::AnalyzerRegistry;
use crate::analysis::config::AnalysisConfig;
use crate::core::{LuciError, Result};
use crate::mapping::Mapping;

use crate::query::ast::ScoringExpression;
use crate::query::parser::opt_str;
use crate::search::{MissingValue, SortField, SortFieldType, SortOrder, SortValue};
use crate::storage::SingleFileDirectory;

use crate::search::expression::SearchExpression;

use crate::query::Query as _;
use crate::reader::IndexReader;
use crate::search::searcher::Searcher;
use crate::writer::IndexWriter;

/// Parsed contents of the `user_metadata` blob written by
/// [`IndexWriter::commit`].
struct ParsedUserMetadata {
    mapping: Mapping,
    deletions: crate::deletion::DeletionMap,
}

/// Decode the `user_metadata` blob.
///
/// The blob holds only mapping + deletions; per-field vector indexes
/// live in their own file extents managed by the storage layer (see
/// [[global-vector-indices]]). The format is either
/// length-prefixed `[mapping_len][mapping_json][deletion_bytes]` or,
/// for the oldest indexes, raw mapping JSON with no deletions.
fn parse_user_metadata(meta: &[u8]) -> Result<ParsedUserMetadata> {
    if meta.is_empty() {
        return Ok(ParsedUserMetadata {
            mapping: Mapping::builder().build(),
            deletions: crate::deletion::DeletionMap::new(),
        });
    }
    if meta.first() == Some(&b'{') {
        let json: serde_json::Value = serde_json::from_slice(meta)
            .map_err(|e| LuciError::IndexCorrupted(format!("invalid mapping metadata: {e}")))?;
        return Ok(ParsedUserMetadata {
            mapping: Mapping::from_json(&json)?,
            deletions: crate::deletion::DeletionMap::new(),
        });
    }
    if meta.len() < 4 {
        return Ok(ParsedUserMetadata {
            mapping: Mapping::builder().build(),
            deletions: crate::deletion::DeletionMap::new(),
        });
    }
    let mapping_len = u32::from_le_bytes(meta[0..4].try_into().unwrap()) as usize;
    let mapping_bytes = &meta[4..4 + mapping_len];
    let json: serde_json::Value = serde_json::from_slice(mapping_bytes)
        .map_err(|e| LuciError::IndexCorrupted(format!("invalid mapping metadata: {e}")))?;
    let mapping = Mapping::from_json(&json)?;
    let after_mapping = 4 + mapping_len;
    let deletions = if after_mapping >= meta.len() {
        crate::deletion::DeletionMap::new()
    } else if meta.len() >= after_mapping + 4 {
        // New format: deletion_bytes is length-prefixed.
        let del_len =
            u32::from_le_bytes(meta[after_mapping..after_mapping + 4].try_into().unwrap()) as usize;
        let start = after_mapping + 4;
        if start + del_len <= meta.len() {
            crate::deletion::DeletionMap::from_bytes(&meta[start..start + del_len])?
        } else {
            // Older format wrote deletion bytes raw, no length prefix.
            crate::deletion::DeletionMap::from_bytes(&meta[after_mapping..])?
        }
    } else {
        crate::deletion::DeletionMap::from_bytes(&meta[after_mapping..])?
    };
    Ok(ParsedUserMetadata { mapping, deletions })
}

/// Parse `sort` from request JSON into sort field specs.
///
/// Returns `Ok(None)` only when the `sort` key is absent. A sort value
/// of an unsupported type (e.g. a number), an unknown order/missing
/// string, or an unknown per-field option returns `Err(InvalidQuery)`
/// rather than silently dropping the sort — see
/// [[fix-strict-search-parsing]] and [[code-must-not-lie]].
pub fn parse_sort(
    value: Option<&serde_json::Value>,
) -> crate::core::Result<Option<Vec<crate::search::SortField>>> {
    let arr = match value {
        None => return Ok(None),
        Some(serde_json::Value::Array(a)) => a,
        Some(serde_json::Value::String(s)) => {
            return Ok(Some(vec![parse_sort_item(&serde_json::Value::String(
                s.clone(),
            ))?]));
        }
        Some(obj @ serde_json::Value::Object(_)) => {
            return Ok(Some(vec![parse_sort_item(obj)?]));
        }
        Some(other) => {
            return Err(LuciError::InvalidQuery(format!(
                "sort: must be a string, object, or array of those; got {other}"
            )));
        }
    };

    let items: std::result::Result<Vec<_>, _> = arr.iter().map(parse_sort_item).collect();
    items.map(Some)
}

fn parse_sort_item(value: &serde_json::Value) -> crate::core::Result<crate::search::SortField> {
    match value {
        serde_json::Value::String(s) => {
            let (field, default_order) = match s.as_str() {
                "_score" => (SortFieldType::Score, SortOrder::Desc),
                "_doc" => (SortFieldType::Doc, SortOrder::Asc),
                name => (SortFieldType::Field(name.to_string()), SortOrder::Asc),
            };
            Ok(SortField {
                field,
                order: default_order,
                missing: MissingValue::Last,
            })
        }
        serde_json::Value::Object(obj) => {
            let (name, spec) = obj.iter().next().ok_or_else(|| {
                crate::core::LuciError::InvalidQuery("sort: entry must have a field name".into())
            })?;
            let (field, default_order) = match name.as_str() {
                "_score" => (SortFieldType::Score, SortOrder::Desc),
                "_doc" => (SortFieldType::Doc, SortOrder::Asc),
                name => (SortFieldType::Field(name.to_string()), SortOrder::Asc),
            };
            let (order, missing) = match spec {
                serde_json::Value::String(o) => {
                    let ord = match o.as_str() {
                        "asc" => SortOrder::Asc,
                        "desc" => SortOrder::Desc,
                        other => {
                            return Err(LuciError::InvalidQuery(format!(
                                "sort[{name}]: unknown order '{other}', expected 'asc' or 'desc'"
                            )));
                        }
                    };
                    (ord, MissingValue::Last)
                }
                serde_json::Value::Object(_) => {
                    let ctx = format!("sort[{name}]");
                    let opts = crate::search::expression::validate_obj_keys(
                        spec,
                        &["order", "missing"],
                        &ctx,
                    )?;
                    let ord = match opt_str(opts, "order", &ctx)? {
                        Some("asc") => SortOrder::Asc,
                        Some("desc") => SortOrder::Desc,
                        Some(other) => {
                            return Err(LuciError::InvalidQuery(format!(
                                "{ctx}: unknown order '{other}', expected 'asc' or 'desc'"
                            )));
                        }
                        None => default_order,
                    };
                    let miss = match opt_str(opts, "missing", &ctx)? {
                        Some("_first") => MissingValue::First,
                        Some("_last") => MissingValue::Last,
                        Some(other) => {
                            return Err(LuciError::InvalidQuery(format!(
                                "{ctx}: unknown missing '{other}', expected '_first' or '_last'"
                            )));
                        }
                        None => MissingValue::Last,
                    };
                    (ord, miss)
                }
                _ => {
                    return Err(LuciError::InvalidQuery(format!(
                        "sort[{name}]: spec must be \"asc\"/\"desc\" or an object, got {spec}"
                    )));
                }
            };
            Ok(SortField {
                field,
                order,
                missing,
            })
        }
        _ => Err(LuciError::InvalidQuery(format!(
            "sort: entry must be a field-name string or a {{field: spec}} object, got {value}"
        ))),
    }
}

/// Extract inner_hit specs from the query expression tree.
pub(crate) fn extract_inner_hit_specs(
    ast: &ScoringExpression,
    searcher: &crate::search::searcher::Searcher,
) -> crate::core::Result<Vec<crate::query::nested::InnerHitSpec>> {
    let mut specs = Vec::new();
    collect_inner_hit_specs(ast, searcher, &mut specs)?;
    Ok(specs)
}

fn collect_inner_hit_specs(
    ast: &ScoringExpression,
    searcher: &crate::search::searcher::Searcher,
    specs: &mut Vec<crate::query::nested::InnerHitSpec>,
) -> crate::core::Result<()> {
    match ast {
        ScoringExpression::Nested {
            path,
            query,
            inner_hits: Some(config),
        } => {
            // Bind a separate weight for inner hit resolution. A bind error
            // (e.g. a bad-field knn under the nested query) must surface, not
            // be swallowed into a missing inner_hits block. See
            // [[feature-knn-query-type]] §4b.
            let weight = query.bind(searcher, crate::core::ScoreMode::Complete)?;
            let name = config.name.clone().unwrap_or_else(|| path.clone());
            specs.push(crate::query::nested::InnerHitSpec {
                name,
                path: path.clone(),
                config: config.clone(),
                weight,
            });
        }
        ScoringExpression::Bool {
            must,
            should,
            must_not,
            filter,
            ..
        } => {
            for sub in must.iter().chain(should).chain(must_not).chain(filter) {
                collect_inner_hit_specs(sub, searcher, specs)?;
            }
        }
        ScoringExpression::Nested { query, .. } => {
            collect_inner_hit_specs(query, searcher, specs)?;
        }
        _ => {}
    }
    Ok(())
}

/// Parse `search_after` cursor values from request JSON.
pub fn parse_search_after(
    value: Option<&serde_json::Value>,
) -> crate::core::Result<Option<Vec<crate::search::SortValue>>> {
    let arr = match value {
        None | Some(serde_json::Value::Null) => return Ok(None),
        Some(serde_json::Value::Array(a)) => a,
        Some(other) => {
            return Err(LuciError::InvalidQuery(format!(
                "search_after: must be an array of cursor values, got {other}"
            )));
        }
    };
    let values = arr
        .iter()
        .map(|v| match v {
            serde_json::Value::Number(n) => {
                if let Some(i) = n.as_i64() {
                    Ok(SortValue::I64(i))
                } else if let Some(f) = n.as_f64() {
                    Ok(SortValue::F64(f))
                } else {
                    Err(LuciError::InvalidQuery(format!(
                        "search_after: numeric cursor value out of range: {n}"
                    )))
                }
            }
            serde_json::Value::String(s) => Ok(SortValue::Str(s.clone())),
            serde_json::Value::Bool(b) => Ok(SortValue::Bool(*b)),
            serde_json::Value::Null => Ok(SortValue::Null),
            other => Err(LuciError::InvalidQuery(format!(
                "search_after: cursor values must be a number, string, boolean, or null; \
                 got {other}"
            ))),
        })
        .collect::<crate::core::Result<Vec<_>>>()?;
    Ok(Some(values))
}

/// Parse `_source` from request JSON into a `SourceFilter`.
pub fn parse_source_filter(value: Option<&serde_json::Value>) -> crate::search::SourceFilter {
    use crate::search::SourceFilter;
    match value {
        None | Some(serde_json::Value::Bool(true)) => SourceFilter::Enabled,
        Some(serde_json::Value::Bool(false)) => SourceFilter::Disabled,
        Some(serde_json::Value::String(s)) => SourceFilter::Fields(vec![s.clone()]),
        Some(serde_json::Value::Array(arr)) => {
            let fields: Vec<String> = arr
                .iter()
                .filter_map(|v| v.as_str().map(String::from))
                .collect();
            SourceFilter::Fields(fields)
        }
        Some(serde_json::Value::Object(obj)) => {
            let includes = obj
                .get("includes")
                .and_then(|v| v.as_array())
                .map(|arr| {
                    arr.iter()
                        .filter_map(|v| v.as_str().map(String::from))
                        .collect()
                })
                .unwrap_or_default();
            let excludes = obj
                .get("excludes")
                .and_then(|v| v.as_array())
                .map(|arr| {
                    arr.iter()
                        .filter_map(|v| v.as_str().map(String::from))
                        .collect()
                })
                .unwrap_or_default();
            SourceFilter::IncludeExclude { includes, excludes }
        }
        _ => SourceFilter::Enabled,
    }
}

/// A Luci search index.
///
/// Provides a unified API for indexing and searching JSON documents.
/// Owns the storage, schema, writer, and search state.
pub struct Index {
    schema: Mapping,
    analysis_config: Option<AnalysisConfig>,
    /// Writer state — behind Mutex for exclusive write access.
    writer: std::sync::Mutex<WriterState>,
    /// Reader state — behind RwLock for concurrent read access.
    reader: std::sync::RwLock<ReaderState>,
    /// Shared file handle for reader refresh. Avoids opening a new fd,
    /// which would break `fcntl` lock semantics.
    /// See [[architecture-cross-process-locking#Critical Constraint]].
    file_handle: std::sync::Arc<std::fs::File>,
    /// Transaction state: true when a transaction is active.
    /// Protected by `txn_mutex` + `txn_condvar` for blocking.
    /// See [[architecture-concurrency-transactions]].
    txn_mutex: std::sync::Mutex<bool>,
    txn_condvar: std::sync::Condvar,
}

struct WriterState {
    writer: IndexWriter,
    commit_generation: u64,
}

struct ReaderState {
    /// Committed segment state. `Arc`-wrapped so `SearchResults` can
    /// hold a reference that outlives the next commit.
    segment_store: Option<std::sync::Arc<crate::search::segment_store::SegmentStore>>,
    store_generation: u64,
}

impl Index {
    /// Create a new index at the given path with fully dynamic mappings.
    ///
    /// All field types are inferred from the first document that contains each
    /// field. Mappings are persisted to disk on commit.
    pub fn create(path: impl AsRef<Path>) -> Result<Self> {
        Self::create_with_mapping(path, Mapping::builder().build())
    }

    /// Create a new index with explicit field mappings.
    ///
    /// Unknown fields are handled according to the mapping's `dynamic` mode
    /// (default: `true`, meaning auto-map and index).
    pub fn create_with_mapping(path: impl AsRef<Path>, mapping: Mapping) -> Result<Self> {
        Self::create_with_settings(path, mapping, None)
    }

    /// Create a new index with explicit mappings and analysis settings.
    ///
    /// The `analysis_config` defines custom analyzers, tokenizers, char filters,
    /// and token filters. It is persisted in the index metadata so that
    /// [`open`](Self::open) can rebuild the analyzer registry.
    ///
    /// See [[feature-analysis-pipeline]].
    pub fn create_with_settings(
        path: impl AsRef<Path>,
        mut mapping: Mapping,
        analysis_config: Option<AnalysisConfig>,
    ) -> Result<Self> {
        // Validate cross-field references (e.g., copy_to targets must
        // exist in the schema) before any work is done. Catches
        // builder-API users who skipped Mapping::from_json — see
        // [[code-must-not-lie]].
        mapping.validate()?;
        mapping.ensure_id_field();
        let path = path.as_ref().to_path_buf();
        let storage = SingleFileDirectory::create(&path)?;
        let file_handle = storage.file_handle();
        let writer_analyzers = match &analysis_config {
            Some(config) => config
                .build_registry()
                .map_err(|e| LuciError::InvalidQuery(e))?,
            None => AnalyzerRegistry::new(),
        };
        let mut writer = IndexWriter::new(storage, mapping.clone(), writer_analyzers);
        // Persist analysis config so open() can rebuild the registry.
        if let Some(ref config) = analysis_config {
            writer.set_analysis_json(Some(config.to_json()));
        }
        Ok(Self {
            schema: mapping,
            analysis_config,
            writer: std::sync::Mutex::new(WriterState {
                writer,
                commit_generation: 0,
            }),
            reader: std::sync::RwLock::new(ReaderState {
                segment_store: None,
                store_generation: 0,
            }),
            file_handle,
            txn_mutex: std::sync::Mutex::new(false),
            txn_condvar: std::sync::Condvar::new(),
        })
    }

    /// Open an existing index at the given path.
    ///
    /// Mappings are loaded from disk (persisted on each commit). If the index
    /// has no persisted mappings (pre-persistence format), an empty mapping
    /// with `dynamic=true` is used.
    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
        let path = path.as_ref().to_path_buf();
        let storage = SingleFileDirectory::open(&path)?;
        let generation = storage.generation();

        // Load persisted mapping + deletions + global HNSW from user
        // metadata. The framing dispatch lives in `parse_user_metadata`
        // — see [[global-vector-indices]] for the v2 format.
        let meta = storage.user_metadata();
        let parsed = parse_user_metadata(meta)?;
        let mut mapping = parsed.mapping;
        mapping.ensure_id_field();

        // Reconstruct the global HNSW from per-field vector indexes
        // managed by the storage layer (one extent per field).
        let global_hnsw = crate::vector::global::GlobalHnsw::new(&mapping);
        for field_id in storage.vector_index_fields() {
            let Some(bytes) = storage.read_vector_index(field_id)? else {
                continue;
            };
            global_hnsw.load_field(field_id, &bytes)?;
        }

        // Load analysis config from settings if present
        let analysis_config = Self::load_analysis_config_from_storage(&storage);
        let writer_analyzers = match &analysis_config {
            Some(config) => config
                .build_registry()
                .unwrap_or_else(|_| AnalyzerRegistry::new()),
            None => AnalyzerRegistry::new(),
        };
        let file_handle = storage.file_handle();
        let mut writer = IndexWriter::new(storage, mapping.clone(), writer_analyzers);
        if let Some(ref config) = analysis_config {
            writer.set_analysis_json(Some(config.to_json()));
        }
        writer.load_deletions(parsed.deletions);
        writer.load_global_hnsw(global_hnsw);
        Ok(Self {
            schema: mapping,
            analysis_config,
            writer: std::sync::Mutex::new(WriterState {
                writer,
                commit_generation: generation,
            }),
            reader: std::sync::RwLock::new(ReaderState {
                segment_store: None,
                store_generation: 0,
            }),
            file_handle,
            txn_mutex: std::sync::Mutex::new(false),
            txn_condvar: std::sync::Condvar::new(),
        })
    }

    /// Load analysis config from storage metadata.
    ///
    /// Looks for a `"settings"."analysis"` key in the persisted mapping JSON.
    fn load_analysis_config_from_storage(storage: &SingleFileDirectory) -> Option<AnalysisConfig> {
        let meta = storage.user_metadata();
        if meta.is_empty() {
            return None;
        }
        let parsed = parse_user_metadata(meta).ok()?;
        let json = parsed.mapping.to_json();
        let analysis = json.get("settings")?.get("analysis")?;
        AnalysisConfig::from_json(analysis).ok()
    }

    /// Add a document. Auto-commits — the document is immediately searchable.
    ///
    /// Blocks if a transaction is active (waits for it to complete).
    /// See [[architecture-concurrency-transactions]].
    pub fn add(&self, doc: serde_json::Value) -> Result<()> {
        self.wait_for_transaction();
        let mut w = self.writer.lock().unwrap();
        w.writer.add(doc)?;
        self.commit_inner(&mut w)
    }

    /// Add multiple documents in a single call. Auto-commits at the end —
    /// all documents are searchable after this returns.
    ///
    /// Blocks if a transaction is active (waits for it to complete).
    /// Reduces Python→Rust FFI overhead from one call per document to one
    /// call per batch. Fails fast on the first invalid document.
    ///
    /// Returns `{"took": <ms>, "count": <n>}`.
    ///
    /// See [[feature-bulk-api]].
    pub fn bulk(&self, docs: Vec<serde_json::Value>) -> Result<serde_json::Value> {
        self.wait_for_transaction();
        let start = std::time::Instant::now();
        let total = docs.len();
        let mut w = self.writer.lock().unwrap();
        for (i, doc) in docs.into_iter().enumerate() {
            w.writer
                .add(doc)
                .map_err(|e| LuciError::InvalidQuery(format!("bulk item {i}: {e}")))?;
        }
        self.commit_inner(&mut w)?;
        Ok(serde_json::json!({
            "took": start.elapsed().as_millis() as u64,
            "count": total
        }))
    }

    /// Get a document by its `_id`. Returns the source if found.
    /// See [[feature-document-crud]].
    pub fn get(&self, id: &str) -> Result<Option<serde_json::Value>> {
        let expr = SearchExpression::from_json(serde_json::json!({"term": {"_id": id}}), 10)?;
        let results = self.search(&expr)?;
        Ok(results.hit(0).and_then(|h| h.source()))
    }

    /// Delete a document by its `_id`. Returns true if found and deleted.
    /// See [[feature-document-crud]].
    pub fn delete(&self, id: &str) -> Result<bool> {
        // delete() auto-commits below, so block until any active transaction
        // completes (consistent with add()/bulk()) — otherwise the commit
        // would prematurely flush the transaction's buffered documents.
        self.wait_for_transaction();
        let expr = SearchExpression::from_json(serde_json::json!({"term": {"_id": id}}), 1)?;
        let results = self.search(&expr)?;
        if let Some(hit) = results.hit(0) {
            let mut w = self.writer.lock().unwrap();
            w.writer.mark_deleted(hit.segment_id(), hit.doc_id());
            // Auto-commit so the deletion is durable on return, matching
            // add()/bulk(). Without this a bare delete() is lost on process
            // exit. See luci-index/tests/deletion_persistence.rs.
            self.commit_inner(&mut w)?;
            return Ok(true);
        }
        Ok(false)
    }

    /// Update a document by its `_id` with a partial doc merge.
    /// Returns true if found and updated. See [[feature-document-crud]].
    pub fn update(&self, id: &str, partial_doc: serde_json::Value) -> Result<bool> {
        // Auto-commits below; wait out any active transaction (see delete()).
        self.wait_for_transaction();
        let expr = SearchExpression::from_json(serde_json::json!({"term": {"_id": id}}), 10)?;
        let results = self.search(&expr)?;
        let (seg_id, doc_id, mut source) = match results.hit(0) {
            Some(hit) => match hit.source() {
                Some(s) => (hit.segment_id(), hit.doc_id(), s),
                None => return Ok(false),
            },
            None => return Ok(false),
        };

        // Merge partial doc into existing source
        if let (Some(existing_obj), Some(partial_obj)) =
            (source.as_object_mut(), partial_doc.as_object())
        {
            for (k, v) in partial_obj {
                existing_obj.insert(k.clone(), v.clone());
            }
        }

        // Preserve the _id
        if let Some(obj) = source.as_object_mut() {
            obj.insert("_id".to_string(), serde_json::Value::String(id.to_string()));
        }

        // Replace atomically: mark the old doc deleted and add the new version
        // in a single commit, so a crash can't leave the doc deleted but not
        // re-added (which separate self.delete()+self.add() would now risk,
        // since delete() auto-commits).
        let mut w = self.writer.lock().unwrap();
        w.writer.mark_deleted(seg_id, doc_id);
        w.writer.add(source)?;
        self.commit_inner(&mut w)?;
        Ok(true)
    }

    /// Delete all documents matching a query. Returns count deleted.
    /// See [[feature-document-crud]].
    pub fn delete_by_query(&self, query: serde_json::Value) -> Result<u64> {
        // Auto-commits below; wait out any active transaction (see delete()).
        self.wait_for_transaction();
        let expr = SearchExpression::from_json(
            serde_json::json!({"query": query, "size": 10000, "_source": false}),
            10000,
        )?;
        let results = self.search(&expr)?;
        let count = results.len() as u64;
        let mut w = self.writer.lock().unwrap();
        for hit in results.iter() {
            w.writer.mark_deleted(hit.segment_id(), hit.doc_id());
        }
        // Commit once after marking the whole batch (durable on return,
        // matching add()/bulk()). Skip the commit when nothing matched.
        if count > 0 {
            self.commit_inner(&mut w)?;
        }
        Ok(count)
    }

    /// Count documents matching a query.
    /// See [[feature-document-crud]].
    pub fn count(&self, query: serde_json::Value) -> Result<u64> {
        let expr = SearchExpression::from_json(
            serde_json::json!({"query": query, "_source": false, "size": 100000}),
            100000,
        )?;
        let results = self.search(&expr)?;
        Ok(results.len() as u64)
    }

    /// Flush and commit buffered documents. Internal — called automatically
    /// by `add()` and `bulk()`. Not part of the public API.
    /// Takes a mutable reference to WriterState (caller holds the Mutex).
    fn commit_inner(&self, w: &mut WriterState) -> Result<()> {
        w.writer.commit()?;
        w.commit_generation += 1;
        Ok(())
    }

    /// Ensure the reader's segment store is up-to-date with the latest commit.
    /// If the reader is stale, rebuilds it under a write lock.
    fn refresh_reader(&self) -> Result<()> {
        let commit_gen = self.writer.lock().unwrap().commit_generation;
        let store_gen = self.reader.read().unwrap().store_generation;

        if store_gen == commit_gen && self.reader.read().unwrap().segment_store.is_some() {
            return Ok(());
        }

        // Stale — rebuild using the shared file handle to avoid
        // opening a new fd (which would break fcntl lock semantics).
        let storage = SingleFileDirectory::open_from_handle(self.file_handle.clone())?;
        let reader = IndexReader::open(&storage)?;
        let store_analyzers = match &self.analysis_config {
            Some(config) => config
                .build_registry()
                .unwrap_or_else(|_| AnalyzerRegistry::new()),
            None => AnalyzerRegistry::new(),
        };

        // Snapshot the global HNSW from the same commit the writer just
        // produced. Reading from storage rather than sharing an `Arc`
        // with the writer keeps the read-side view frozen until the
        // next commit, matching the existing reader-isolation semantics
        // of `SegmentReader`.
        let global_hnsw = {
            let g = crate::vector::global::GlobalHnsw::new(&self.schema);
            for field_id in storage.vector_index_fields() {
                if let Some(bytes) = storage.read_vector_index(field_id)? {
                    g.load_field(field_id, &bytes)?;
                }
            }
            Some(std::sync::Arc::new(g))
        };

        let store = crate::search::segment_store::SegmentStore::new(
            reader.into_segments(),
            store_analyzers,
            Some(self.schema.clone()),
            global_hnsw,
        );

        let mut r = self.reader.write().unwrap();
        r.segment_store = Some(std::sync::Arc::new(store));
        r.store_generation = commit_gen;
        Ok(())
    }

    /// Force-merge all segments down to at most `max_segments`.
    ///
    /// Call after bulk indexing is complete to optimize for query performance.
    /// Invalidates the cached searcher.
    pub fn force_merge(&self, max_segments: usize) -> Result<()> {
        let mut w = self.writer.lock().unwrap();
        w.writer.force_merge(max_segments)?;
        w.commit_generation += 1;
        Ok(())
    }

    /// Search the index with a native SearchExpression.
    ///
    /// Thin entry point: refreshes the reader, delegates to
    /// `Searcher::execute_query`, then handles deletion filtering
    /// and total_hits capping.
    pub fn search(
        &self,
        expr: &crate::search::expression::SearchExpression,
    ) -> Result<crate::search::results::SearchResults> {
        use crate::query::ast::QueryExpression;

        // Ensure reader is up-to-date with the latest commit
        self.refresh_reader()?;

        let r = self.reader.read().unwrap();
        let store = r.segment_store.as_ref().unwrap();
        let searcher = Searcher::new(store);

        // Resolve the query — default to match_all
        let match_all = QueryExpression::Scoring(ScoringExpression::MatchAll);
        let query = expr.query.as_ref().unwrap_or(&match_all);

        // Searcher handles all dispatch (scoring vs ranking). Errors
        // from bind / scorer_supplier / scorer propagate — see
        // [[fix-silent-scorer-errors]].
        let mut results = searcher.execute_query(query, expr)?;

        // Apply rescore
        if let Some(ref rescore) = expr.rescore {
            searcher.apply_rescore(
                &mut results,
                rescore.query.as_ref(),
                rescore.window_size,
                rescore.query_weight,
                rescore.rescore_query_weight,
                rescore.score_mode,
            )?;
        }

        // Clone the Arc<SegmentStore> for the SearchResults, then drop the
        // reader lock before acquiring the writer lock for deletion filtering.
        // This maintains the lock ordering rule: writer before reader.
        let store = r.segment_store.as_ref().unwrap().clone();
        drop(r);

        // Filter out deleted documents
        let w = self.writer.lock().unwrap();
        let deletions = w.writer.deletions();
        let pre_len = results.hits.len() as u64;
        results
            .hits
            .retain(|hit| !deletions.is_deleted(hit.segment_id, hit.doc_id));
        let removed = pre_len - results.hits.len() as u64;
        drop(w);

        // Adjust total_hits
        if removed > 0 || results.total_hits.value > results.hits.len() as u64 {
            results.total_hits =
                crate::search::TotalHits::exact(results.total_hits.value.saturating_sub(removed));
        }

        // Apply track_total_hits capping
        results.total_hits =
            crate::search::TotalHits::resolve(results.total_hits.value, expr.track_total_hits);

        // Wrap ScoringResults with SegmentStore for lazy Hit access
        Ok(crate::search::results::SearchResults::new(
            results.hits,
            results.total_hits,
            results.aggregations,
            store,
            expr.query.clone(),
        ))
    }

    /// Set the memory budget for auto-flush (bytes).
    ///
    /// When the in-memory buffer exceeds this size, it is automatically
    /// flushed to a new segment. Smaller values produce more segments
    /// (better parallelism, more merge work). Default: 64 MB.
    pub fn set_memory_budget(&self, budget: usize) {
        self.writer.lock().unwrap().writer.set_memory_budget(budget);
    }

    /// Set the timeout for acquiring the cross-process write lock.
    ///
    /// Default: 5 seconds. If another process holds the write lock,
    /// retries until the timeout, then returns `WriterLocked`.
    /// Not persisted — applies only to this session.
    pub fn set_write_timeout(&self, timeout: std::time::Duration) {
        self.writer
            .lock()
            .unwrap()
            .writer
            .set_write_timeout(timeout);
    }

    /// Get the schema.
    pub fn schema(&self) -> &Mapping {
        &self.schema
    }

    /// Number of buffered (uncommitted) documents.
    pub fn buffered_doc_count(&self) -> u32 {
        self.writer.lock().unwrap().writer.buffered_doc_count()
    }

    // --- Transaction API ---
    // These methods support the Python Transaction context manager.
    // See [[architecture-concurrency-transactions]].

    /// Block until no transaction is active.
    ///
    /// Called by `add()` and `bulk()` to wait for any active transaction
    /// to complete before proceeding with auto-commit writes.
    fn wait_for_transaction(&self) {
        let guard = self.txn_mutex.lock().unwrap();
        // Drop the guard when the condition is met (txn not active).
        let _guard = self
            .txn_condvar
            .wait_while(guard, |active| *active)
            .unwrap();
    }

    /// Begin a transaction. Blocks if another transaction is active.
    ///
    /// While a transaction is active, `add()` and `bulk()` will block
    /// until the transaction completes.
    pub fn begin_transaction(&self) -> Result<()> {
        let mut active = self.txn_mutex.lock().unwrap();
        // Wait for any existing transaction to finish.
        active = self.txn_condvar.wait_while(active, |a| *a).unwrap();
        *active = true;
        Ok(())
    }

    /// End a transaction, waking any blocked writers.
    ///
    /// Must only be called after a successful `begin_transaction()`.
    pub fn end_transaction(&self) {
        let mut active = self.txn_mutex.lock().unwrap();
        *active = false;
        self.txn_condvar.notify_all();
    }

    /// Whether a transaction is currently active.
    pub fn is_transaction_active(&self) -> bool {
        *self.txn_mutex.lock().unwrap()
    }

    /// Add a document without committing. For use inside a transaction.
    ///
    /// The caller must have called `begin_transaction()` first.
    pub fn txn_add(&self, doc: serde_json::Value) -> Result<()> {
        let mut w = self.writer.lock().unwrap();
        w.writer.add(doc)?;
        Ok(())
    }

    /// Commit all buffered documents. For use inside a transaction.
    ///
    /// The caller must have called `begin_transaction()` first.
    pub fn txn_commit(&self) -> Result<()> {
        let mut w = self.writer.lock().unwrap();
        self.commit_inner(&mut w)
    }

    /// Discard all buffered (uncommitted) documents. For transaction rollback.
    ///
    /// Resets the writer's in-memory buffer without flushing to storage.
    pub fn txn_rollback(&self) {
        let mut w = self.writer.lock().unwrap();
        w.writer.discard_buffer();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::mapping::FieldType;
    use serde_json::json;

    fn test_dir(name: &str) -> std::path::PathBuf {
        let dir =
            std::env::temp_dir().join(format!("luci_index_facade_{}_{name}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        dir
    }

    fn cleanup(path: &Path) {
        let _ = std::fs::remove_dir_all(path);
    }

    fn test_schema() -> Mapping {
        Mapping::builder()
            .field("title", FieldType::Text)
            .field("body", FieldType::Text)
            .field("status", FieldType::Keyword)
            .build()
    }

    #[test]
    fn create_add_commit_search() {
        let path = test_dir("basic");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();

        index
            .add(json!({"title": "Hello World", "body": "A greeting", "status": "published"}))
            .unwrap();
        index
            .add(json!({"title": "Search Engine", "body": "Building search", "status": "draft"}))
            .unwrap();

        let results = index
            .search(&SearchExpression::from_json(json!({"match": {"title": "hello"}}), 10).unwrap())
            .unwrap();
        assert_eq!(results.total_hits().value, 1);
        assert_eq!(
            results.hit(0).unwrap().source().unwrap()["title"],
            "Hello World"
        );

        cleanup(&path);
    }

    #[test]
    fn search_with_query_wrapper() {
        let path = test_dir("wrapper");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();

        index
            .add(json!({"title": "Rust Programming", "status": "published"}))
            .unwrap();
        index
            .add(json!({"title": "Go Programming", "status": "published"}))
            .unwrap();

        let results = index
            .search(
                &SearchExpression::from_json(json!({"query": {"match": {"title": "rust"}}}), 10)
                    .unwrap(),
            )
            .unwrap();
        assert_eq!(results.total_hits().value, 1);

        cleanup(&path);
    }

    #[test]
    fn term_query_on_keyword() {
        let path = test_dir("keyword");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();

        index
            .add(json!({"title": "A", "status": "published"}))
            .unwrap();
        index.add(json!({"title": "B", "status": "draft"})).unwrap();
        index
            .add(json!({"title": "C", "status": "published"}))
            .unwrap();

        let results = index
            .search(
                &SearchExpression::from_json(json!({"term": {"status": "published"}}), 10).unwrap(),
            )
            .unwrap();
        assert_eq!(results.total_hits().value, 2);

        cleanup(&path);
    }

    #[test]
    fn bool_query_end_to_end() {
        let path = test_dir("bool");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();

        index
            .add(json!({"title": "Search Engine Design", "status": "published"}))
            .unwrap();
        index
            .add(json!({"title": "Search Tips", "status": "draft"}))
            .unwrap();
        index
            .add(json!({"title": "Database Design", "status": "published"}))
            .unwrap();

        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "bool": {
                            "must": [{"match": {"title": "search"}}],
                            "filter": [{"term": {"status": "published"}}]
                        }
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();

        assert_eq!(results.total_hits().value, 1);
        let title = results.hit(0).unwrap().source().unwrap()["title"]
            .as_str()
            .unwrap()
            .to_string();
        assert!(title.contains("Search Engine"));

        cleanup(&path);
    }

    #[test]
    fn phrase_query_end_to_end() {
        let path = test_dir("phrase");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();

        index
            .add(json!({"body": "the quick brown fox jumps"}))
            .unwrap();
        index.add(json!({"body": "brown quick fox"})).unwrap();

        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({"match_phrase": {"body": "quick brown fox"}}),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(results.total_hits().value, 1);

        cleanup(&path);
    }

    #[test]
    fn match_all_query() {
        let path = test_dir("match_all");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();

        index.add(json!({"title": "A"})).unwrap();
        index.add(json!({"title": "B"})).unwrap();
        index.add(json!({"title": "C"})).unwrap();

        let results = index
            .search(&SearchExpression::from_json(json!({"match_all": {}}), 10).unwrap())
            .unwrap();
        assert_eq!(results.total_hits().value, 3);

        cleanup(&path);
    }

    #[test]
    fn open_existing_index() {
        let path = test_dir("reopen");

        {
            let index = Index::create_with_mapping(&path, test_schema()).unwrap();
            index
                .add(json!({"title": "persistent doc", "status": "published"}))
                .unwrap();
        }

        {
            let index = Index::open(&path).unwrap();
            let results = index
                .search(
                    &SearchExpression::from_json(json!({"match": {"title": "persistent"}}), 10)
                        .unwrap(),
                )
                .unwrap();
            assert_eq!(results.total_hits().value, 1);
        }

        cleanup(&path);
    }

    #[test]
    fn multiple_commits_searchable() {
        let path = test_dir("multi_commit");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();

        index.add(json!({"title": "first batch"})).unwrap();
        index.add(json!({"title": "second batch"})).unwrap();

        let results = index
            .search(&SearchExpression::from_json(json!({"match": {"title": "batch"}}), 10).unwrap())
            .unwrap();
        assert_eq!(results.total_hits().value, 2);

        cleanup(&path);
    }

    #[test]
    fn empty_index_search() {
        let path = test_dir("empty");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();

        let results = index
            .search(&SearchExpression::from_json(json!({"match_all": {}}), 10).unwrap())
            .unwrap();
        assert_eq!(results.total_hits().value, 0);

        cleanup(&path);
    }

    #[test]
    fn search_no_results() {
        let path = test_dir("no_results");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();
        index.add(json!({"title": "hello"})).unwrap();

        let results = index
            .search(
                &SearchExpression::from_json(json!({"term": {"status": "nonexistent"}}), 10)
                    .unwrap(),
            )
            .unwrap();
        assert_eq!(results.total_hits().value, 0);

        cleanup(&path);
    }

    #[test]
    fn constant_score_end_to_end() {
        let path = test_dir("const_score");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();
        index
            .add(json!({"title": "test", "status": "published"}))
            .unwrap();

        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "constant_score": {
                            "filter": {"term": {"status": "published"}},
                            "boost": 3.14
                        }
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(results.total_hits().value, 1);
        assert!((results.hit(0).unwrap().score() - 3.14).abs() < 0.01);

        cleanup(&path);
    }

    #[test]
    fn from_offset_pagination() {
        let path = test_dir("from_offset");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();

        // Index 20 docs with distinct titles (bulk to get one segment)
        let docs: Vec<_> = (0..20)
            .map(|i| json!({"title": format!("document {i}"), "status": "published"}))
            .collect();
        index.bulk(docs).unwrap();

        // Page 1: from=0, size=5
        let page1 = index
            .search(
                &SearchExpression::from_json(
                    json!({"query": {"match_all": {}}, "from": 0, "size": 5}),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(page1.len(), 5);
        assert_eq!(page1.total_hits().value, 20);

        // Page 2: from=5, size=5
        let page2 = index
            .search(
                &SearchExpression::from_json(
                    json!({"query": {"match_all": {}}, "from": 5, "size": 5}),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(page2.len(), 5);
        assert_eq!(page2.total_hits().value, 20);

        // Pages should have no overlap
        let page1_ids: Vec<_> = page1.iter().map(|h| h.doc_id()).collect();
        let page2_ids: Vec<_> = page2.iter().map(|h| h.doc_id()).collect();
        for id in &page2_ids {
            assert!(!page1_ids.contains(id), "page 2 should not overlap page 1");
        }

        // from beyond total_hits → empty
        let empty = index
            .search(
                &SearchExpression::from_json(
                    json!({"query": {"match_all": {}}, "from": 100, "size": 5}),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(empty.len(), 0);
        assert_eq!(empty.total_hits().value, 20);

        // from=0 should match default behavior
        let default = index
            .search(&SearchExpression::from_json(json!({"match_all": {}}), 5).unwrap())
            .unwrap();
        let explicit = index
            .search(
                &SearchExpression::from_json(
                    json!({"query": {"match_all": {}}, "from": 0, "size": 5}),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(default.len(), explicit.len());

        cleanup(&path);
    }

    #[test]
    fn source_filtering() {
        let path = test_dir("source_filter");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();

        index
            .add(json!({"title": "Hello World", "body": "Full text", "status": "published"}))
            .unwrap();
        index
            .add(json!({"title": "Search Engine", "body": "More text", "status": "draft"}))
            .unwrap();

        // _source: false — no source at all
        let sf_disabled = crate::search::SourceFilter::Disabled;
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({"query": {"match_all": {}}, "_source": false}),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(results.len(), 2);
        assert!(
            results
                .hit(0)
                .unwrap()
                .source_filtered(&sf_disabled)
                .is_none()
        );
        assert!(
            results
                .hit(1)
                .unwrap()
                .source_filtered(&sf_disabled)
                .is_none()
        );

        // _source: ["title"] — only title field
        let sf_title = crate::search::SourceFilter::Fields(vec!["title".to_string()]);
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({"query": {"match_all": {}}, "_source": ["title"]}),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        for hit in results.iter() {
            let src = hit.source_filtered(&sf_title).unwrap();
            assert!(src.get("title").is_some());
            assert!(src.get("body").is_none());
            assert!(src.get("status").is_none());
        }

        // _source: {"excludes": ["body"]} — all except body
        let sf_excl = crate::search::SourceFilter::IncludeExclude {
            includes: vec![],
            excludes: vec!["body".to_string()],
        };
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "_source": {"excludes": ["body"]}
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        for hit in results.iter() {
            let src = hit.source_filtered(&sf_excl).unwrap();
            assert!(src.get("title").is_some());
            assert!(src.get("status").is_some());
            assert!(src.get("body").is_none());
        }

        // _source: {"includes": ["title", "body"], "excludes": ["body"]} — exclude wins
        let sf_incl_excl = crate::search::SourceFilter::IncludeExclude {
            includes: vec!["title".to_string(), "body".to_string()],
            excludes: vec!["body".to_string()],
        };
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "_source": {"includes": ["title", "body"], "excludes": ["body"]}
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        for hit in results.iter() {
            let src = hit.source_filtered(&sf_incl_excl).unwrap();
            assert!(src.get("title").is_some());
            assert!(src.get("body").is_none());
        }

        // _source: true — full source (default)
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({"query": {"match_all": {}}, "_source": true}),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        for hit in results.iter() {
            let src = hit.source().unwrap();
            assert!(src.get("title").is_some());
            assert!(src.get("body").is_some());
            assert!(src.get("status").is_some());
        }

        cleanup(&path);
    }

    #[test]
    fn fields_retrieval() {
        let path = test_dir("fields_retrieval");
        let schema = Mapping::builder()
            .field("title", FieldType::Text)
            .field("tag", FieldType::Keyword)
            .field("price", FieldType::Float)
            .build();
        let index = Index::create_with_mapping(&path, schema).unwrap();

        index
            .add(json!({"title": "Hello World", "tag": "tech", "price": 9.99}))
            .unwrap();
        index
            .add(json!({"title": "Search Engine", "tag": "science", "price": 19.99}))
            .unwrap();

        // Retrieve keyword + numeric fields
        let field_names: Vec<String> = vec!["tag".to_string(), "price".to_string()];
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "fields": ["tag", "price"],
                        "_source": false
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();

        assert_eq!(results.len(), 2);
        let sf_disabled = crate::search::SourceFilter::Disabled;
        for hit in results.iter() {
            // _source disabled
            assert!(hit.source_filtered(&sf_disabled).is_none());
            // fields present
            let fields = hit.fields(&field_names);
            assert!(fields.get("tag").is_some(), "tag field should be present");
            assert!(
                fields.get("price").is_some(),
                "price field should be present"
            );
            // Values are arrays (ES compat)
            let tag = fields.get("tag").unwrap();
            assert!(tag.is_array(), "tag should be array");
        }

        // Text field silently omitted (no columnar store)
        let title_fields: Vec<String> = vec!["title".to_string()];
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "fields": ["title"],
                        "_source": false
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        for hit in results.iter() {
            let fields = hit.fields(&title_fields);
            assert!(
                fields.get("title").is_none(),
                "text field should be omitted"
            );
        }

        // Fields + source together
        let tag_fields: Vec<String> = vec!["tag".to_string()];
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "fields": ["tag"],
                        "_source": ["title"]
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        for hit in results.iter() {
            assert!(hit.source().is_some());
            assert!(!hit.fields(&tag_fields).is_empty());
        }

        cleanup(&path);
    }

    #[test]
    fn sort_by_field() {
        let path = test_dir("sort_by_field");
        let schema = Mapping::builder()
            .field("title", FieldType::Text)
            .field("tag", FieldType::Keyword)
            .field("price", FieldType::Float)
            .build();
        let index = Index::create_with_mapping(&path, schema).unwrap();

        index
            .add(json!({"title": "Expensive", "tag": "b", "price": 99.99}))
            .unwrap();
        index
            .add(json!({"title": "Cheap", "tag": "a", "price": 1.99}))
            .unwrap();
        index
            .add(json!({"title": "Mid", "tag": "c", "price": 49.99}))
            .unwrap();

        // Sort by price ascending
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "sort": ["price"]
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(results.len(), 3);
        let prices: Vec<f64> = results
            .iter()
            .map(|h| h.sort_values().unwrap()[0].to_json().as_f64().unwrap())
            .collect();
        assert!(
            prices[0] <= prices[1] && prices[1] <= prices[2],
            "prices should be ascending: {:?}",
            prices
        );

        // Sort by price descending
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "sort": [{"price": "desc"}]
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        let prices: Vec<f64> = results
            .iter()
            .map(|h| h.sort_values().unwrap()[0].to_json().as_f64().unwrap())
            .collect();
        assert!(
            prices[0] >= prices[1] && prices[1] >= prices[2],
            "prices should be descending: {:?}",
            prices
        );

        // Sort by keyword field
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "sort": [{"tag": "asc"}]
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        let tags: Vec<String> = results
            .iter()
            .map(|h| {
                h.sort_values().unwrap()[0]
                    .to_json()
                    .as_str()
                    .unwrap()
                    .to_string()
            })
            .collect();
        assert_eq!(tags, vec!["a", "b", "c"]);

        // Sort values present in response
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "sort": ["price"]
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        for hit in results.iter() {
            assert!(hit.sort_values().is_some(), "sort values should be present");
        }

        // Sort by _score (should match default ordering)
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "sort": ["_score"]
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(results.len(), 3);

        cleanup(&path);
    }

    #[test]
    fn search_after_pagination() {
        let path = test_dir("search_after");
        let schema = Mapping::builder()
            .field("title", FieldType::Text)
            .field("price", FieldType::Float)
            .build();
        let index = Index::create_with_mapping(&path, schema).unwrap();

        let docs: Vec<_> = (0..20)
            .map(|i| json!({"title": format!("item {i}"), "price": i as f64}))
            .collect();
        index.bulk(docs).unwrap();

        // Page 1: sort by price, size=5
        let page1 = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "sort": ["price"],
                        "size": 5
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(page1.len(), 5);
        let last_hit = page1.hit(page1.len() - 1).unwrap();
        let last_sort = last_hit.sort_values().unwrap();

        // Page 2: search_after with last sort values
        let page2 = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "sort": ["price"],
                        "size": 5,
                        "search_after": [last_sort[0].to_json()]
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(page2.len(), 5);

        // No overlap between pages
        let p1_prices: Vec<f64> = page1
            .iter()
            .map(|h| h.sort_values().unwrap()[0].to_json().as_f64().unwrap())
            .collect();
        let p2_prices: Vec<f64> = page2
            .iter()
            .map(|h| h.sort_values().unwrap()[0].to_json().as_f64().unwrap())
            .collect();
        assert!(
            p1_prices.last().unwrap() < p2_prices.first().unwrap(),
            "page 2 should start after page 1: {:?} vs {:?}",
            p1_prices,
            p2_prices
        );

        // Iterate all pages
        let mut all_prices = Vec::new();
        let mut cursor: Option<Vec<serde_json::Value>> = None;
        loop {
            let mut req = json!({
                "query": {"match_all": {}},
                "sort": ["price"],
                "size": 7
            });
            if let Some(ref c) = cursor {
                req["search_after"] = serde_json::json!(c);
            }
            let page = index
                .search(&SearchExpression::from_json(req, 10).unwrap())
                .unwrap();
            if page.is_empty() {
                break;
            }
            for hit in page.iter() {
                all_prices.push(hit.sort_values().unwrap()[0].to_json().as_f64().unwrap());
            }
            let last = page.hit(page.len() - 1).unwrap();
            cursor = Some(
                last.sort_values()
                    .unwrap()
                    .iter()
                    .map(|sv| sv.to_json())
                    .collect(),
            );
        }
        assert_eq!(all_prices.len(), 20, "should iterate all 20 docs");
        // Verify ascending order and no duplicates
        for i in 1..all_prices.len() {
            assert!(
                all_prices[i] > all_prices[i - 1],
                "should be strictly ascending: {} vs {}",
                all_prices[i - 1],
                all_prices[i]
            );
        }

        // search_after beyond all results → empty
        let empty = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "sort": ["price"],
                        "search_after": [999.0]
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert!(empty.is_empty());

        cleanup(&path);
    }

    #[test]
    fn explain_score() {
        let path = test_dir("explain");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();

        index
            .bulk(vec![
                json!({"title": "hello world", "status": "published"}),
                json!({"title": "hello hello hello", "status": "draft"}),
            ])
            .unwrap();

        // Explain is lazy — always available when a query is provided
        let results = index
            .search(
                &SearchExpression::from_json(json!({"query": {"match": {"title": "hello"}}}), 10)
                    .unwrap(),
            )
            .unwrap();
        assert_eq!(results.len(), 2);
        for hit in results.iter() {
            let exp = hit
                .explain()
                .expect("explain should not error")
                .expect("explain should be present");
            assert!(exp.value > 0.0, "score should be positive");
            assert!(
                !exp.description.is_empty(),
                "description should be non-empty"
            );
            assert!(!exp.details.is_empty(), "should have BM25 sub-details");
        }

        // Higher TF should have higher explanation value
        let scores: Vec<f32> = results
            .iter()
            .map(|h| h.explain().unwrap().unwrap().value)
            .collect();
        // hits are score-descending, so first should have higher explain value
        assert!(scores[0] >= scores[1]);

        cleanup(&path);
    }

    #[test]
    fn collapse_by_field() {
        let path = test_dir("collapse");
        let schema = Mapping::builder()
            .field("title", FieldType::Text)
            .field("author", FieldType::Keyword)
            .build();
        let index = Index::create_with_mapping(&path, schema).unwrap();

        // Multiple docs per author
        index
            .add(json!({"title": "post one by alice", "author": "alice"}))
            .unwrap();
        index
            .add(json!({"title": "post two by alice", "author": "alice"}))
            .unwrap();
        index
            .add(json!({"title": "post by bob", "author": "bob"}))
            .unwrap();
        index
            .add(json!({"title": "another by bob", "author": "bob"}))
            .unwrap();
        index
            .add(json!({"title": "post by carol", "author": "carol"}))
            .unwrap();

        // Without collapse: all 5 docs
        let results = index
            .search(&SearchExpression::from_json(json!({"query": {"match_all": {}}}), 10).unwrap())
            .unwrap();
        assert_eq!(results.len(), 5);

        // With collapse: 1 per author = 3 hits
        let author_fields: Vec<String> = vec!["author".to_string()];
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "collapse": {"field": "author"}
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(results.len(), 3, "should have 3 unique authors");

        // Each hit should have the author field
        let mut authors: Vec<String> = results
            .iter()
            .map(|h| h.fields(&author_fields))
            .filter_map(|f| f.get("author").cloned())
            .filter_map(|v| v.as_array().cloned())
            .filter_map(|a| a.first().cloned())
            .filter_map(|v| v.as_str().map(String::from))
            .collect();
        authors.sort();
        assert_eq!(authors, vec!["alice", "bob", "carol"]);

        // total_hits should be the uncollapsed count
        assert_eq!(results.total_hits().value, 5);

        cleanup(&path);
    }

    #[test]
    fn track_total_hits() {
        use crate::search::TotalHitsRelation;

        let path = test_dir("track_total_hits");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();
        for i in 0..100 {
            index
                .add(json!({"title": format!("document {i}"), "status": "published"}))
                .unwrap();
        }

        // Default: exact count
        let results = index
            .search(&SearchExpression::from_json(json!({"query": {"match_all": {}}}), 10).unwrap())
            .unwrap();
        assert_eq!(results.total_hits().value, 100);
        assert_eq!(results.total_hits().relation, TotalHitsRelation::EqualTo);

        // track_total_hits: true — same as default
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "track_total_hits": true
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(results.total_hits().value, 100);
        assert_eq!(results.total_hits().relation, TotalHitsRelation::EqualTo);

        // track_total_hits: false — disabled
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "track_total_hits": false
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(results.total_hits().value, 0);
        assert_eq!(
            results.total_hits().relation,
            TotalHitsRelation::GreaterThanOrEqualTo
        );

        // track_total_hits: 50 — cap below actual
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "track_total_hits": 50
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(results.total_hits().value, 50);
        assert_eq!(
            results.total_hits().relation,
            TotalHitsRelation::GreaterThanOrEqualTo
        );

        // track_total_hits: 200 — cap above actual
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "track_total_hits": 200
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(results.total_hits().value, 100);
        assert_eq!(results.total_hits().relation, TotalHitsRelation::EqualTo);

        cleanup(&path);
    }

    #[test]
    fn inner_hits_nested() {
        let path = test_dir("inner_hits");
        let schema = Mapping::builder()
            .field("product", FieldType::Text)
            .field("offers", FieldType::Nested)
            .field("offers.seller", FieldType::Keyword)
            .field("offers.price", FieldType::Keyword)
            .build();
        let index = Index::create_with_mapping(&path, schema).unwrap();

        index
            .add(json!({
                "product": "laptop",
                "offers": [
                    {"seller": "Alice", "price": "999"},
                    {"seller": "Bob", "price": "1299"}
                ]
            }))
            .unwrap();

        // Nested query with inner_hits — should return which offer matched
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "nested": {
                            "path": "offers",
                            "query": {
                                "term": {"offers.seller": "Alice"}
                            },
                            "inner_hits": {}
                        }
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();

        assert_eq!(results.total_hits().value, 1);
        assert_eq!(results.len(), 1);

        // Check inner_hits are present
        let hit = results.hit(0).unwrap();
        let inner = hit
            .inner_hits()
            .expect("inner_hits should not error")
            .expect("inner_hits should be present");
        let offers = inner
            .get("offers")
            .expect("should have 'offers' inner_hits group");

        let inner_hits_obj = offers.get("hits").unwrap();
        let inner_total = inner_hits_obj
            .get("total")
            .unwrap()
            .get("value")
            .unwrap()
            .as_u64()
            .unwrap();
        assert_eq!(inner_total, 1, "should have 1 matching inner hit");

        let inner_docs = inner_hits_obj.get("hits").unwrap().as_array().unwrap();
        assert_eq!(inner_docs.len(), 1);

        // The inner hit should have Alice's offer as _source
        let inner_source = inner_docs[0].get("_source").unwrap();
        assert_eq!(inner_source.get("seller").unwrap(), "Alice");
        assert_eq!(inner_source.get("price").unwrap(), "999");

        cleanup(&path);
    }

    #[test]
    fn rescore_reranks() {
        let path = test_dir("rescore");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();

        // Doc 0: has "hello" (matches initial query)
        // Doc 1: has "hello world" (matches both initial and rescore phrase)
        // Doc 2: has "hello there world" (matches initial, weak phrase match)
        index
            .add(json!({"title": "hello", "status": "published"}))
            .unwrap();
        index
            .add(json!({"title": "hello world", "status": "published"}))
            .unwrap();
        index
            .add(json!({"title": "hello there world", "status": "published"}))
            .unwrap();

        // Initial search: match "hello" — all 3 docs match
        let without_rescore = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match": {"title": "hello"}}
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(without_rescore.len(), 3);

        // With rescore: boost docs matching phrase "hello world"
        let with_rescore = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match": {"title": "hello"}},
                        "rescore": {
                            "window_size": 10,
                            "query": {
                                "rescore_query": {"match_phrase": {"title": "hello world"}},
                                "query_weight": 0.5,
                                "rescore_query_weight": 1.5
                            }
                        }
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(with_rescore.len(), 3);

        // Doc with "hello world" should be boosted to top
        // (it matches both the initial match and the rescore phrase)
        let top_source = with_rescore.hit(0).unwrap().source().unwrap();
        assert_eq!(
            top_source["title"], "hello world",
            "phrase match should be boosted to top"
        );

        // Rescore should change scores
        assert_ne!(
            without_rescore.hit(0).unwrap().score(),
            with_rescore.hit(0).unwrap().score(),
            "rescore should modify scores"
        );

        cleanup(&path);
    }

    #[test]
    fn multi_fields() {
        let path = test_dir("multi_fields");
        // title as text + title.raw as keyword
        let mapping_json = json!({
            "properties": {
                "title": {
                    "type": "text",
                    "fields": {
                        "raw": {"type": "keyword"}
                    }
                }
            }
        });
        let schema = Mapping::from_json(&mapping_json).unwrap();
        let index = Index::create_with_mapping(&path, schema).unwrap();

        index.add(json!({"title": "Hello World"})).unwrap();
        index.add(json!({"title": "Hello Luci"})).unwrap();
        index.add(json!({"title": "Goodbye World"})).unwrap();

        // Text search on parent field
        let results = index
            .search(&SearchExpression::from_json(json!({"match": {"title": "hello"}}), 10).unwrap())
            .unwrap();
        assert_eq!(
            results.total_hits().value,
            2,
            "text search should match 2 docs"
        );

        // Exact match on sub-field (keyword)
        let results = index
            .search(
                &SearchExpression::from_json(json!({"term": {"title.raw": "Hello World"}}), 10)
                    .unwrap(),
            )
            .unwrap();
        assert_eq!(
            results.total_hits().value,
            1,
            "exact keyword match on sub-field"
        );

        // Term query on sub-field shouldn't match partial text
        let results = index
            .search(
                &SearchExpression::from_json(json!({"term": {"title.raw": "hello"}}), 10).unwrap(),
            )
            .unwrap();
        assert_eq!(
            results.total_hits().value,
            0,
            "keyword sub-field is not analyzed"
        );

        // Sort by sub-field
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "sort": [{"title.raw": "asc"}]
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert_eq!(results.len(), 3);
        let titles: Vec<String> = results
            .iter()
            .map(|h| {
                h.sort_values().unwrap()[0]
                    .to_json()
                    .as_str()
                    .unwrap()
                    .to_string()
            })
            .collect();
        assert_eq!(titles[0], "Goodbye World");
        assert_eq!(titles[1], "Hello Luci");
        assert_eq!(titles[2], "Hello World");

        // Aggregation on sub-field
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "aggs": {"titles": {"terms": {"field": "title.raw"}}},
                        "size": 0
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();
        assert!(results.aggregations().contains_key("titles"));

        cleanup(&path);
    }

    #[test]
    fn copy_to() {
        let path = test_dir("copy_to");
        let schema = Mapping::from_json(&json!({
            "properties": {
                "title": {"type": "text", "copy_to": "all_text"},
                "body": {"type": "text", "copy_to": "all_text"},
                "tag": {"type": "keyword"},
                "all_text": {"type": "text"}
            }
        }))
        .unwrap();
        let index = Index::create_with_mapping(&path, schema).unwrap();

        index
            .add(json!({"title": "search engine", "body": "fast and embedded", "tag": "tech"}))
            .unwrap();
        index
            .add(json!({"title": "database", "body": "columnar storage", "tag": "tech"}))
            .unwrap();

        // Search on individual field
        let results = index
            .search(
                &SearchExpression::from_json(json!({"match": {"title": "search"}}), 10).unwrap(),
            )
            .unwrap();
        assert_eq!(results.total_hits().value, 1);

        // Search on copy_to target — should find doc with "search" in title
        let results = index
            .search(
                &SearchExpression::from_json(json!({"match": {"all_text": "search"}}), 10).unwrap(),
            )
            .unwrap();
        assert_eq!(
            results.total_hits().value,
            1,
            "copy_to target should contain title content"
        );

        // Search on copy_to target — should find doc with "embedded" in body
        let results = index
            .search(
                &SearchExpression::from_json(json!({"match": {"all_text": "embedded"}}), 10)
                    .unwrap(),
            )
            .unwrap();
        assert_eq!(
            results.total_hits().value,
            1,
            "copy_to target should contain body content"
        );

        // Search on copy_to target — "columnar" from body of doc 2
        let results = index
            .search(
                &SearchExpression::from_json(json!({"match": {"all_text": "columnar"}}), 10)
                    .unwrap(),
            )
            .unwrap();
        assert_eq!(
            results.total_hits().value,
            1,
            "copy_to target should contain body of second doc"
        );

        // all_text should NOT appear in _source
        let results = index
            .search(
                &SearchExpression::from_json(json!({"match": {"all_text": "search"}}), 10).unwrap(),
            )
            .unwrap();
        let src = results.hit(0).unwrap().source().unwrap();
        assert!(
            src.get("all_text").is_none(),
            "copy_to target should not be in _source"
        );

        cleanup(&path);
    }

    #[test]
    fn document_crud() {
        let path = test_dir("crud");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();

        // Add with user-provided _id
        index
            .add(json!({"_id": "doc1", "title": "Hello", "status": "published"}))
            .unwrap();
        index
            .add(json!({"_id": "doc2", "title": "World", "status": "draft"}))
            .unwrap();

        // Get by ID
        let doc = index.get("doc1").unwrap().expect("doc1 should exist");
        assert_eq!(doc["title"], "Hello");

        // Get missing
        assert!(index.get("nonexistent").unwrap().is_none());

        // Count
        assert_eq!(index.count(json!({"match_all": {}})).unwrap(), 2);

        // Delete by ID
        assert!(index.delete("doc1").unwrap());
        assert!(index.get("doc1").unwrap().is_none());
        assert_eq!(index.count(json!({"match_all": {}})).unwrap(), 1);

        // Delete missing
        assert!(!index.delete("nonexistent").unwrap());

        // Update by ID
        index
            .add(json!({"_id": "doc3", "title": "Original", "status": "published"}))
            .unwrap();
        assert!(index.update("doc3", json!({"title": "Updated"})).unwrap());
        let doc = index
            .get("doc3")
            .unwrap()
            .expect("doc3 should exist after update");
        assert_eq!(doc["title"], "Updated");
        assert_eq!(doc["status"], "published"); // preserved

        // Delete by query
        index
            .add(json!({"_id": "d1", "title": "draft one", "status": "draft"}))
            .unwrap();
        index
            .add(json!({"_id": "d2", "title": "draft two", "status": "draft"}))
            .unwrap();
        let deleted = index
            .delete_by_query(json!({"term": {"status": "draft"}}))
            .unwrap();
        assert!(deleted >= 2);

        // Auto-generated IDs
        index.add(json!({"title": "Auto ID doc"})).unwrap();
        let results = index
            .search(&SearchExpression::from_json(json!({"match": {"title": "auto"}}), 1).unwrap())
            .unwrap();
        assert_eq!(results.total_hits().value, 1);

        cleanup(&path);
    }

    #[test]
    fn filter_agg_counts_only_matching_docs() {
        let path = test_dir("filter_agg");
        let schema = Mapping::builder()
            .field("title", FieldType::Text)
            .field("status", FieldType::Keyword)
            .field("price", FieldType::Float)
            .build();
        let index = Index::create_with_mapping(&path, schema).unwrap();

        // 3 active docs, 2 draft docs
        index
            .add(json!({"title": "doc1", "status": "active", "price": 10.0}))
            .unwrap();
        index
            .add(json!({"title": "doc2", "status": "active", "price": 20.0}))
            .unwrap();
        index
            .add(json!({"title": "doc3", "status": "active", "price": 30.0}))
            .unwrap();
        index
            .add(json!({"title": "doc4", "status": "draft", "price": 40.0}))
            .unwrap();
        index
            .add(json!({"title": "doc5", "status": "draft", "price": 50.0}))
            .unwrap();

        // Filter agg: count only active docs
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "aggs": {
                            "active_only": {
                                "filter": {"term": {"status": "active"}}
                            }
                        },
                        "size": 0
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();

        // total_hits should be 5 (all docs match the top-level match_all)
        assert_eq!(results.total_hits().value, 5);

        // filter agg should count only the 3 active docs, NOT all 5
        let aggs = results.aggregations();
        let active = aggs.get("active_only").expect("active_only agg missing");
        let active_json = active.to_json();
        let doc_count = active_json["buckets"][0]["doc_count"]
            .as_u64()
            .expect("filter agg should have doc_count in buckets[0]");
        assert_eq!(
            doc_count, 3,
            "filter agg should count 3 active docs, got {doc_count}"
        );

        cleanup(&path);
    }

    #[test]
    fn value_count_uses_stats_fast_path() {
        let path = test_dir("value_count_fast");
        let schema = Mapping::builder()
            .field("title", FieldType::Text)
            .field("price", FieldType::Float)
            .build();
        let index = Index::create_with_mapping(&path, schema).unwrap();

        for i in 0..5 {
            index
                .add(json!({"title": format!("doc {i}"), "price": (i as f64) * 10.0}))
                .unwrap();
        }

        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({
                        "query": {"match_all": {}},
                        "aggs": {
                            "count_price": {"value_count": {"field": "price"}}
                        },
                        "size": 0
                    }),
                    10,
                )
                .unwrap(),
            )
            .unwrap();

        let aggs = results.aggregations();
        let count_agg = aggs.get("count_price").expect("count_price missing");
        let value = count_agg.to_json()["value"].as_f64().unwrap();
        assert_eq!(value, 5.0, "value_count should be 5");

        cleanup(&path);
    }

    /// Regression test for [[investigation-20260405-04-multi-match-wrong-rewrite]].
    ///
    /// multi_match defaults to best_fields (dis_max), not most_fields
    /// (bool/should). A doc matching in one strong field should outscore
    /// a doc matching weakly in multiple fields.
    #[test]
    fn multi_match_best_fields_scoring() {
        let path = test_dir("multi_match_bf");
        let index = Index::create_with_mapping(&path, test_schema()).unwrap();

        // Doc 0: "search" appears twice in title (high TF → high score), absent from body
        index
            .add(json!({"title": "search search", "body": "web application framework"}))
            .unwrap();
        // Doc 1: "search" once in title AND once in body (lower per-field BM25)
        index
            .add(json!({"title": "database search", "body": "search tools"}))
            .unwrap();

        // multi_match with default best_fields: score = max(field_scores)
        let results = index
            .search(
                &SearchExpression::from_json(
                    json!({"multi_match": {"query": "search", "fields": ["title", "body"]}}),
                    10,
                )
                .unwrap(),
            )
            .unwrap();

        assert_eq!(results.total_hits().value, 2);

        // Also run individual match queries to get per-field scores for doc 1
        let title_results = index
            .search(
                &SearchExpression::from_json(json!({"match": {"title": "search"}}), 10).unwrap(),
            )
            .unwrap();
        let body_results = index
            .search(&SearchExpression::from_json(json!({"match": {"body": "search"}}), 10).unwrap())
            .unwrap();

        // Find doc 1's per-field scores
        let doc1_title_score = title_results
            .iter()
            .find(|h| h.source().unwrap()["body"] == "search tools")
            .map(|h| h.score())
            .unwrap();
        let doc1_body_score = body_results
            .iter()
            .find(|h| h.source().unwrap()["body"] == "search tools")
            .map(|h| h.score())
            .unwrap();
        let doc1_max = doc1_title_score.max(doc1_body_score);
        let doc1_sum = doc1_title_score + doc1_body_score;

        // Find doc 1's multi_match score
        let doc1_mm_score = results
            .iter()
            .find(|h| h.source().unwrap()["body"] == "search tools")
            .map(|h| h.score())
            .unwrap();

        // With best_fields (correct): score ≈ max(field_scores)
        // With bool/should (bug): score ≈ sum(field_scores)
        assert!(
            (doc1_mm_score - doc1_max).abs() < 1e-5,
            "multi_match score ({doc1_mm_score}) should equal max of field scores ({doc1_max}), \
             not sum ({doc1_sum})"
        );

        cleanup(&path);
    }

    /// Regression test for [[fix-disjunction-heap-inefficiency]].
    ///
    /// PrefixQuery should produce constant scores per matching doc, not
    /// `count_of_matching_terms`. A doc matching 2 prefix-terms must
    /// score the same as a doc matching 1 prefix-term.
    #[test]
    fn prefix_query_constant_score() {
        let path = test_dir("prefix_const_score");
        let schema = Mapping::builder().field("body", FieldType::Text).build();
        let index = Index::create_with_mapping(&path, schema).unwrap();

        // Doc 0: contains 2 prefix-matching terms ("search" and "searchable")
        index
            .add(json!({"body": "search searchable engines"}))
            .unwrap();
        // Doc 1: contains 1 prefix-matching term ("searching")
        index.add(json!({"body": "searching for tools"})).unwrap();

        let results = index
            .search(&SearchExpression::from_json(json!({"prefix": {"body": "sear"}}), 10).unwrap())
            .unwrap();
        assert_eq!(results.total_hits().value, 2);

        // Both docs should have identical scores under constant-score semantics.
        let doc0_score = results
            .iter()
            .find(|h| h.source().unwrap()["body"] == "search searchable engines")
            .map(|h| h.score())
            .unwrap();
        let doc1_score = results
            .iter()
            .find(|h| h.source().unwrap()["body"] == "searching for tools")
            .map(|h| h.score())
            .unwrap();

        assert!(
            (doc0_score - doc1_score).abs() < 1e-5,
            "prefix scores must be constant: doc0 ({doc0_score}) vs doc1 ({doc1_score}). \
             ES uses CONSTANT_SCORE_BLENDED_REWRITE — every matching doc gets the same boost \
             regardless of how many prefix-terms match."
        );

        cleanup(&path);
    }

    /// Regression test for [[fix-wildcard-fuzzy-quadratic-dedup]].
    ///
    /// Multi-segment wildcard query must return the union of per-segment
    /// matches. Per-segment rewrite means each segment enumerates its own
    /// terms independently — assert no docs are missed and duplicate
    /// terms across segments don't cause spurious extras.
    #[test]
    fn wildcard_multi_segment_returns_all_matches() {
        let path = test_dir("wildcard_multi_seg");
        let schema = Mapping::builder().field("tag", FieldType::Keyword).build();
        let index = Index::create_with_mapping(&path, schema).unwrap();

        // Each add() auto-commits, creating a segment per call (until merge).
        // The test is correct even if the writer merges eagerly — the
        // wildcard must still find all 5 matching docs.
        index.add(json!({"tag": "tech1"})).unwrap();
        index.add(json!({"tag": "tech2"})).unwrap();
        index.add(json!({"tag": "other"})).unwrap();
        index.add(json!({"tag": "tech3"})).unwrap();
        index.add(json!({"tag": "tech1"})).unwrap(); // duplicate across segments
        index.add(json!({"tag": "tech4"})).unwrap();
        index.add(json!({"tag": "unrelated"})).unwrap();

        let results = index
            .search(
                &SearchExpression::from_json(json!({"wildcard": {"tag": "tech*"}}), 20).unwrap(),
            )
            .unwrap();

        // Expected: 5 docs match (tech1 twice, tech2, tech3, tech4).
        assert_eq!(
            results.total_hits().value,
            5,
            "multi-segment wildcard missed matches"
        );

        cleanup(&path);
    }

    // --- E10: strict value types in sort / search_after. A malformed
    // spec must error, never silently sort-by-score or drop the cursor.
    // See [[code-must-not-lie]] and [[fix-strict-search-parsing]]. ---

    #[test]
    fn parse_sort_unknown_order_rejected() {
        let err = parse_sort(Some(&json!([{"price": {"order": "ascending"}}]))).unwrap_err();
        assert!(format!("{err}").contains("order"), "{err}");
    }

    #[test]
    fn parse_sort_non_string_order_rejected() {
        let err = parse_sort(Some(&json!([{"price": {"order": 1}}]))).unwrap_err();
        assert!(format!("{err}").contains("order"), "{err}");
    }

    #[test]
    fn parse_sort_object_form_accepted() {
        // A single-object sort (ES-valid) now routes through
        // parse_sort_item instead of being silently dropped to no-sort.
        let sort = parse_sort(Some(&json!({"price": "desc"})))
            .unwrap()
            .unwrap();
        assert_eq!(sort.len(), 1);
    }

    #[test]
    fn parse_sort_malformed_entry_rejected() {
        let err = parse_sort(Some(&json!([5]))).unwrap_err();
        assert!(format!("{err}").contains("sort"), "{err}");
    }

    #[test]
    fn parse_search_after_object_element_rejected() {
        let err = parse_search_after(Some(&json!([1, {}]))).unwrap_err();
        assert!(format!("{err}").contains("search_after"), "{err}");
    }

    #[test]
    fn parse_search_after_valid_cursor_accepted() {
        let cursor = parse_search_after(Some(&json!([9.99, "doc-42"])))
            .unwrap()
            .unwrap();
        assert_eq!(cursor.len(), 2);
    }
}