//! Integration tests for the ranked search cascade and the streaming search
//! service, against real temp databases.

use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use quicksearch_core::db::open_or_recreate;
use quicksearch_core::db::repo::{insert_file, set_content_done, NewFile};
use quicksearch_core::mime::FileType;
use quicksearch_core::query::split::split_for_cascade;
use quicksearch_core::search::{
    cascade, MatchField, SearchHit, SearchOptions, SearchService, SearchUpdate,
};
use quicksearch_core::testutil::{zstd_of, Scratch};

struct Seeder {
    conn: rusqlite::Connection,
    store_text: bool,
}

impl Seeder {
    fn new(path: &Path, store_text: bool) -> Seeder {
        Seeder {
            conn: open_or_recreate(path.to_str().unwrap(), "trigram").unwrap(),
            store_text,
        }
    }

    /// Insert a file; `text: Some(..)` also content-indexes it. `dir` has no
    /// trailing separator; the stored parent always carries one.
    fn add(&mut self, name: &str, dir: &str, mtime: u64, text: Option<&str>) -> i64 {
        let parent = format!("{}/", dir);
        let tx = self.conn.transaction().unwrap();
        let id = insert_file(
            &tx,
            &NewFile {
                name,
                parent: &parent,
                size: 42,
                mtime,
                mime: Some("text/plain"),
                ftype: FileType::TEXT,
                hash: None,
                needs_content: true,
            },
        )
        .unwrap()
        .expect("unique path");
        if let Some(text) = text {
            let zstd = self.store_text.then(|| zstd_of(text)).flatten();
            set_content_done(&tx, id, text, zstd.as_deref()).unwrap();
        }
        tx.commit().unwrap();
        id
    }

    fn done(self) -> rusqlite::Connection {
        self.conn
    }
}

/// Run a search and return its hits in rank order — arrival order is scan
/// order, not rank order. Use [`run_collect_batches`] for the stream itself.
fn run_collect(
    conn: &rusqlite::Connection,
    input: &str,
    options: &SearchOptions,
) -> (Vec<SearchHit>, cascade::Outcome) {
    let (batches, outcome) = run_collect_batches(conn, input, options);
    let mut hits: Vec<SearchHit> = batches.into_iter().flatten().collect();
    hits.sort_by(|a, b| {
        a.rank
            .partial_cmp(&b.rank)
            .unwrap_or(std::cmp::Ordering::Equal)
            .then_with(|| a.name.cmp(&b.name))
            .then_with(|| a.path.cmp(&b.path))
    });
    (hits, outcome)
}

/// The raw batch stream, one `Vec` per sink call.
fn run_collect_batches(
    conn: &rusqlite::Connection,
    input: &str,
    options: &SearchOptions,
) -> (Vec<Vec<SearchHit>>, cascade::Outcome) {
    let split = split_for_cascade(input).expect("split");
    let latest = AtomicU64::new(7);
    let mut batches = Vec::new();
    let outcome = cascade::run(conn, &split, options, 7, &latest, &mut |batch| {
        batches.push(batch)
    })
    .expect("cascade run")
    .expect("not cancelled");
    (batches, outcome)
}

fn fuzzy_options() -> SearchOptions {
    SearchOptions {
        fuzzy: true,
        ..SearchOptions::default()
    }
}

fn fuzzy_options_with_edits(max_edits: usize) -> SearchOptions {
    SearchOptions {
        fuzzy: true,
        fuzzy_max_edits: max_edits,
        ..SearchOptions::default()
    }
}

#[test]
fn rank_classification_across_all_stages() {
    let (_dir, p) = Scratch::db("ranks");
    let mut s = Seeder::new(&p, true);
    let rank1 = s.add("Report", "/a", 1, None);
    let rank2 = s.add("report", "/b", 2, None);
    let rank3 = s.add("Quarterly_Report.txt", "/c", 3, None);
    let rank4 = s.add("quarterly_report.txt", "/d", 4, None);
    let rank5 = s.add("notes-cs.txt", "/e", 5, Some("the Report was filed today"));
    let rank6 = s.add("notes-ci.txt", "/f", 6, Some("the report was filed today"));
    let rank7 = s.add("Reprot.txt", "/g", 7, None); // 2 substitutions
    let rank8 = s.add("body-fuzzy.txt", "/h", 8, Some("the reoprt went missing"));
    let rank9 = s.add("alpha.bin", "/Report-archive", 9, None);
    let rank10 = s.add("beta.bin", "/report-archive", 10, None);
    let rank11 = s.add("gamma.bin", "/Reprot-archive", 11, None);
    let _miss = s.add("unrelated.bin", "/z", 12, Some("nothing to see"));
    let conn = s.done();

    let (hits, outcome) = run_collect(&conn, "Report", &fuzzy_options());
    let order: Vec<(i64, u8)> = hits.iter().map(|h| (h.file_id, h.stage)).collect();
    assert_eq!(
        order,
        vec![
            (rank1, 1),
            (rank2, 2),
            (rank3, 3),
            (rank4, 4),
            (rank5, 5),
            (rank6, 6),
            (rank7, 7),
            (rank8, 8),
            (rank9, 9),
            (rank10, 10),
            (rank11, 11),
        ],
        "each fixture lands at its designed rank, in order"
    );
    assert_eq!(outcome.total, 11);
    assert!(!outcome.limited);

    for pair in hits.windows(2) {
        assert!(
            pair[0].rank <= pair[1].rank,
            "ranks must never decrease: {} then {}",
            pair[0].rank,
            pair[1].rank
        );
    }

    for h in hits
        .iter()
        .filter(|h| h.stage == 5 || h.stage == 6 || h.stage == 8)
    {
        let snip = h.snippet.as_ref().expect("full-text hit has a snippet");
        for &(a, b) in &snip.ranges {
            assert!(a < b && b <= snip.window.len());
        }
    }
}

/// Found through the `parent LIKE` half of the one-piece prefilter — the
/// other half of the boundary story told by
/// [`a_wildcard_spanning_the_directory_boundary_is_still_found`].
#[test]
fn path_substring_tiers_split_by_case() {
    let (_dir, p) = Scratch::db("pathcase");
    let mut s = Seeder::new(&p, true);
    let exact = s.add("a.bin", "/srv/Vacation/raw", 1, None);
    let anycase = s.add("b.bin", "/vacation-2023", 2, None);
    let _miss = s.add("c.bin", "/holiday", 3, None);
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "Vacation", &SearchOptions::default());
    assert_eq!(
        hits.iter()
            .map(|h| (h.file_id, h.stage))
            .collect::<Vec<_>>(),
        vec![(exact, 9), (anycase, 10)],
        "exact-case path matches outrank any-case ones"
    );

    let hit = &hits[0];
    let snip = hit.snippet.as_ref().expect("path hits mark the match");
    assert_eq!(snip.window, hit.path, "the window is the whole path");
    assert_eq!(snip.ranges.len(), 1);
    let (a, b) = snip.ranges[0];
    assert_eq!(&snip.window[a..b], "Vacation");
    assert!(!snip.truncated_start && !snip.truncated_end);
}

#[test]
fn path_match_is_deduped_against_a_better_name_match() {
    let (_dir, p) = Scratch::db("pathdedup");
    let mut s = Seeder::new(&p, true);
    let star = s.add("Budget.txt", "/Budget/2024", 1, None);
    let conn = s.done();

    let (hits, outcome) = run_collect(&conn, "Budget", &fuzzy_options());
    assert_eq!(outcome.total, 1);
    assert_eq!(hits.len(), 1);
    assert_eq!(hits[0].file_id, star);
    assert_eq!(hits[0].stage, 3, "the name tier wins");
}

#[test]
fn short_terms_respect_the_three_char_trigram_floor() {
    let (_dir, p) = Scratch::db("pathfloor");
    let mut s = Seeder::new(&p, true);
    let dir_only = s.add("z.bin", "/abcdir", 1, None);
    let name_hit = s.add("ab.txt", "/d", 2, None);
    let _text_only = s.add("body.txt", "/d", 3, Some("ab ab ab"));
    let conn = s.done();

    let (short, _) = run_collect(&conn, "ab", &fuzzy_options());
    assert_eq!(
        short.iter().map(|h| h.file_id).collect::<Vec<_>>(),
        vec![name_hit],
        "2-char term: filename stages only (text, path and fuzzy all skipped)"
    );

    // Fuzzy is off here, so `ab.txt` (1 edit from `abc`) stays out of it.
    let (long, _) = run_collect(&conn, "abc", &SearchOptions::default());
    assert_eq!(
        long.iter()
            .map(|h| (h.file_id, h.stage))
            .collect::<Vec<_>>(),
        vec![(dir_only, 9)],
        "3-char term: the directory match surfaces"
    );
}

#[test]
fn term_with_separator_matches_across_the_path() {
    let (_dir, p) = Scratch::db("pathsep");
    let mut s = Seeder::new(&p, true);
    let nested = s.add("report-final.txt", "/home/docs", 1, None);
    let _elsewhere = s.add("report-final.txt", "/home/other", 2, None);
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "docs/report", &SearchOptions::default());
    assert_eq!(
        hits.iter()
            .map(|h| (h.file_id, h.stage))
            .collect::<Vec<_>>(),
        vec![(nested, 9)],
        "a term spanning a separator can only match the full path"
    );
}

#[test]
fn fuzzy_path_tier_requires_the_fuzzy_flag() {
    let (_dir, p) = Scratch::db("fuzzypath");
    let mut s = Seeder::new(&p, true);
    let typo_dir = s.add("gamma.bin", "/Reprot-archive", 1, None);
    let conn = s.done();

    let (off, _) = run_collect(&conn, "Report", &SearchOptions::default());
    assert!(off.is_empty(), "no fuzzy stages without the flag");

    let (on, _) = run_collect(&conn, "Report", &fuzzy_options());
    assert_eq!(
        on.iter().map(|h| (h.file_id, h.stage)).collect::<Vec<_>>(),
        vec![(typo_dir, 11)]
    );
    assert!((on[0].rank - 11.2).abs() < 1e-9, "2 edits adds 0.2");
}

#[test]
fn fuzzy_max_edits_widens_and_narrows_the_budget() {
    let (_dir, p) = Scratch::db("fuzzybudget");
    let mut s = Seeder::new(&p, true);
    let two_edits = s.add("quartrely.txt", "/d", 1, None);
    let three_edits = s.add("quxxxerly.txt", "/d", 2, None);
    let conn = s.done();

    let (default, _) = run_collect(&conn, "quarterly", &fuzzy_options());
    assert_eq!(
        default.iter().map(|h| h.file_id).collect::<Vec<_>>(),
        vec![two_edits],
        "the default budget of 2 can't reach a 3-edit typo"
    );
    assert!((default[0].rank - 7.2).abs() < 1e-9);

    let (widened, _) = run_collect(&conn, "quarterly", &fuzzy_options_with_edits(3));
    assert_eq!(
        widened.iter().map(|h| h.file_id).collect::<Vec<_>>(),
        vec![two_edits, three_edits],
        "raising the cap admits the 3-edit typo, ranked after the closer one"
    );
    assert!((widened[1].rank - 7.3).abs() < 1e-9);

    let (strict, _) = run_collect(&conn, "quarterly", &fuzzy_options_with_edits(1));
    assert!(strict.is_empty(), "a cap of 1 rejects both typos");

    let (off, _) = run_collect(&conn, "quarterly", &fuzzy_options_with_edits(0));
    assert!(off.is_empty(), "a cap of 0 disables the fuzzy stages");
}

/// `7.0 + 0.1 * distance` reaches 8.0 — the fuzzy *full-text* tier — at ten
/// edits, so a distant filename hit read as a match on the file's contents.
#[test]
fn a_distant_fuzzy_filename_hit_stays_a_name_hit() {
    let (_dir, p) = Scratch::db("fuzzystage");
    let mut s = Seeder::new(&p, true);
    // Ten substitutions against a 30-character term, whose budget is ten.
    let far = s.add("abcdefghijklmnopqrst##########", "/d", 1, None);
    let conn = s.done();

    let (hits, _) = run_collect(
        &conn,
        "abcdefghijklmnopqrstuvwxyz0123",
        &fuzzy_options_with_edits(10),
    );
    assert_eq!(
        hits.iter().map(|h| h.file_id).collect::<Vec<_>>(),
        vec![far]
    );
    assert!(
        (hits[0].rank - 8.0).abs() < 1e-9,
        "rank {} is not the 8.0 that used to truncate into the next stage",
        hits[0].rank
    );
    assert_eq!(hits[0].stage, 7, "the name tier is stage 7 at any distance");
    assert_eq!(hits[0].match_field(), MatchField::Name);
}

#[test]
fn dedup_keeps_best_rank() {
    let (_dir, p) = Scratch::db("dedup");
    let mut s = Seeder::new(&p, true);
    let star = s.add("Budget", "/a", 1, Some("Budget Budget Budget"));
    let conn = s.done();

    let (hits, outcome) = run_collect(&conn, "Budget", &fuzzy_options());
    assert_eq!(outcome.total, 1);
    assert_eq!(hits.len(), 1);
    assert_eq!(hits[0].file_id, star);
    assert_eq!(hits[0].stage, 1);
}

#[test]
fn occurrence_counts_order_within_rank() {
    let (_dir, p) = Scratch::db("frac");
    let mut s = Seeder::new(&p, true);
    let one = s.add("one.txt", "/d", 1, Some("zebra"));
    let three = s.add("three.txt", "/d", 2, Some("zebra zebra zebra"));
    let thousand = s.add("thousand.txt", "/d", 3, Some(&"zebra ".repeat(1500)));
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "zebra", &SearchOptions::default());
    let ids: Vec<i64> = hits.iter().map(|h| h.file_id).collect();
    assert_eq!(
        ids,
        vec![thousand, three, one],
        "more occurrences sorts earlier within the rank"
    );
    assert_eq!(hits[0].rank, 5.0, "1000+ occurrences adds zero");
    assert!((hits[1].rank - 5.997).abs() < 1e-9);
    assert!((hits[2].rank - 5.999).abs() < 1e-9);
}

#[test]
fn like_metacharacters_are_literal() {
    let (_dir, p) = Scratch::db("like");
    let mut s = Seeder::new(&p, true);
    let percent = s.add("100%.txt", "/d", 1, None);
    let underscore = s.add("100_.txt", "/d", 2, None);
    let contains = s.add("x100y.txt", "/d", 3, None);
    // No "100" at all: only leaked `%`/`_` semantics could admit it.
    let _decoy = s.add("1x0y.txt", "/d", 4, None);
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "100%", &SearchOptions::default());
    assert_eq!(
        hits.iter().map(|h| h.file_id).collect::<Vec<_>>(),
        vec![percent],
        "% in the term must not act as a wildcard"
    );

    let (hits, _) = run_collect(&conn, "100*", &SearchOptions::default());
    let mut ids: Vec<i64> = hits.iter().map(|h| h.file_id).collect();
    ids.sort();
    assert_eq!(
        ids,
        vec![percent, underscore, contains],
        "star globs, % and _ stay literal"
    );
}

#[test]
fn diacritic_fts_candidates_are_dropped() {
    let (_dir, p) = Scratch::db("diacritic");
    let mut s = Seeder::new(&p, true);
    // trigram remove_diacritics 1 makes this an FTS candidate for "cafe",
    // but the exact bytes never occur — exact full-text must drop it.
    let _accented = s.add("menu.txt", "/d", 1, Some("le café est ouvert"));
    let plain = s.add("plain.txt", "/d", 2, Some("the cafe is open"));
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "cafe", &SearchOptions::default());
    assert_eq!(
        hits.iter().map(|h| h.file_id).collect::<Vec<_>>(),
        vec![plain]
    );
}

#[test]
fn contentless_mode_degrades_to_unranked_stage6() {
    let (_dir, p) = Scratch::db("notext");
    let mut s = Seeder::new(&p, false); // store_text_for_snippets = false
    let doc = s.add("doc.txt", "/d", 1, Some("walrus columns"));
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "walrus", &fuzzy_options());
    assert_eq!(hits.len(), 1, "FTS still finds it");
    let h = &hits[0];
    assert_eq!(h.file_id, doc);
    assert_eq!(h.stage, 6, "cannot case-verify without text");
    assert!((h.rank - 6.999).abs() < 1e-9, "count-unknown fraction");
    assert!(h.snippet.is_none());
    assert!(!hits.iter().any(|h| h.stage == 8));

    // Without stored text a floor-clearing wildcard can't be pattern-verified.
    let (hits, _) = run_collect(&conn, "wal*rus", &SearchOptions::default());
    assert_eq!(
        hits.iter()
            .map(|h| (h.file_id, h.stage))
            .collect::<Vec<_>>(),
        vec![(doc, 6)]
    );
    assert!(hits[0].snippet.is_none());

    // The short-segment fallback has no FTS evidence, so it finds nothing.
    let (none, _) = run_collect(&conn, "wa*us", &SearchOptions::default());
    assert!(none.is_empty());
}

#[test]
fn filters_apply_to_every_stage() {
    let (_dir, p) = Scratch::db("filters");
    let mut s = Seeder::new(&p, true);
    let keep_name = s.add("alpha.txt", "/keep", 1, None);
    let _skip_name = s.add("alpha.txt", "/skip", 1, None);
    let keep_text = s.add("k.txt", "/keep", 2, Some("alpha body"));
    let _skip_text = s.add("s.txt", "/skip", 2, Some("alpha body"));
    let keep_fuzzy = s.add("alpah.txt", "/keep", 3, None);
    let _skip_fuzzy = s.add("alpah.txt", "/skip", 3, None);
    let keep_path = s.add("p.bin", "/keep/alpha-sub", 4, None);
    let _skip_path = s.add("p.bin", "/skip/alpha-sub", 4, None);
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "alpha path:/keep", &fuzzy_options());
    let mut ids: Vec<i64> = hits.iter().map(|h| h.file_id).collect();
    ids.sort();
    let mut want = vec![keep_name, keep_text, keep_fuzzy, keep_path];
    want.sort();
    assert_eq!(ids, want, "the path filter must gate all stages");
}

/// `flush_pass` is the authority on `limited`: it sets the flag when it
/// actually truncates. `remaining()` reaches zero at the pass boundary
/// whether or not anything was dropped, and the outer loop used to set
/// `limited` there — "(truncated…)" over a set that had dropped nothing.
///
/// The mid-pass boundary is deliberately *not* fixed: a limit filled by the
/// first hits of a pass breaks the scan with rows unexamined, and
/// `cut_short` reports that honestly as "there may be more", which may not
/// be so. Distinguishing the two costs a row scanned past the limit in
/// every pass that fills it — not worth paying on every keystroke.
///
/// The break sits *after* classification: the filename pass's SQL is a
/// superset feeding two rank tiers, so a scanned row is not yet a match,
/// and breaking on one would lose the hits below it.
#[test]
fn the_limit_truncates_flags_and_keeps_the_best_hits() {
    struct Case {
        tag: &'static str,
        subs: usize,
        exact_last: bool,
        limit: usize,
        expect_len: usize,
        expect_limited: bool,
        expect_first: &'static str,
        why: &'static str,
    }
    let cases = [
        Case {
            tag: "limit",
            subs: 10,
            exact_last: false,
            limit: 3,
            expect_len: 3,
            expect_limited: true,
            expect_first: "zz-match-00.txt",
            why: "10 matches under a limit of 3 is a cut set",
        },
        Case {
            // Exactly `limit` results is a complete answer: nothing truncated.
            tag: "limit-exact",
            subs: 3,
            exact_last: false,
            limit: 3,
            expect_len: 3,
            expect_limited: false,
            expect_first: "zz-match-00.txt",
            why: "3 matches under a limit of 3 dropped nothing",
        },
        Case {
            // The known not-fixed boundary: the scan breaks with rows
            // unexamined and `cut_short` says "there may be more".
            tag: "limit-one",
            subs: 1,
            exact_last: false,
            limit: 1,
            expect_len: 1,
            expect_limited: true,
            expect_first: "zz-match-00.txt",
            why: "a scan that stopped with rows unexamined says so",
        },
        Case {
            // The exact match, seeded last, must survive the break.
            tag: "limit-break",
            subs: 20,
            exact_last: true,
            limit: 2,
            expect_len: 2,
            expect_limited: true,
            expect_first: "match",
            why: "21 matches under a limit of 2 is a cut set",
        },
    ];

    for case in &cases {
        let (_dir, p) = Scratch::db(case.tag);
        let mut s = Seeder::new(&p, true);
        for i in 0..case.subs {
            s.add(&format!("zz-match-{:02}.txt", i), "/d", 1, None);
        }
        if case.exact_last {
            s.add("match", "/d", 1, None);
        }
        let conn = s.done();

        let options = SearchOptions {
            limit: case.limit,
            ..SearchOptions::default()
        };
        let (hits, outcome) = run_collect(&conn, "match", &options);
        assert_eq!(hits.len(), case.expect_len, "{}", case.tag);
        assert_eq!(outcome.total, case.expect_len, "{}", case.tag);
        assert_eq!(
            outcome.limited, case.expect_limited,
            "{}: {}",
            case.tag, case.why
        );
        assert_eq!(hits[0].name, case.expect_first, "{}", case.tag);
    }
}

#[test]
fn session_ignores_hide_hits_before_the_cap() {
    let (_dir, p) = Scratch::db("ignores");
    let mut s = Seeder::new(&p, true);
    let keep = s.add("keep-match.txt", "/d", 1, None);
    let _log = s.add("match.log", "/d", 2, None);
    let _sub = s.add("match.txt", "/d/logs", 3, None);
    // Would be a rank-9 path hit, but its parent is an ignored component.
    let _path_hit = s.add("z.bin", "/d/logs/match-sub", 4, None);
    let conn = s.done();

    let options = SearchOptions {
        session_ignores: vec!["*.log".to_string(), "logs".to_string()],
        ..SearchOptions::default()
    };
    let (hits, outcome) = run_collect(&conn, "match", &options);
    assert_eq!(
        hits.iter().map(|h| h.file_id).collect::<Vec<_>>(),
        vec![keep]
    );
    assert_eq!(outcome.total, 1, "ignored rows never count toward totals");
}

#[test]
fn empty_and_filter_only_terms_return_nothing() {
    let (_dir, p) = Scratch::db("empty");
    let mut s = Seeder::new(&p, true);
    s.add("anything.txt", "/d", 1, Some("anything"));
    let conn = s.done();

    for input in ["", "   ", "type:Text"] {
        let (hits, outcome) = run_collect(&conn, input, &SearchOptions::default());
        assert!(hits.is_empty(), "input {:?}", input);
        assert_eq!(outcome.total, 0);
    }
}

#[test]
fn hostile_terms_are_inert() {
    let (_dir, p) = Scratch::db("hostile");
    let mut s = Seeder::new(&p, true);
    s.add("innocent.txt", "/d", 1, Some("innocent content"));
    let conn = s.done();

    for term in [
        "'; DROP TABLE files; --",
        "\" OR 1=1 --",
        // FTS metacharacters after a live wildcard must still be inert.
        "term* (NEAR) : ^",
        "a\0b",
        "*",
        "****",
        "* *",
        &format!("{}*", "x".repeat(10_000)),
    ] {
        let split = split_for_cascade(term).expect("split never fails on words");
        let latest = AtomicU64::new(1);
        let result = cascade::run(
            &conn,
            &split,
            &SearchOptions::default(),
            1,
            &latest,
            &mut |_| {},
        );
        assert!(result.is_ok(), "term {:?}: {:?}", term, result.err());
    }
    let n: i64 = conn
        .query_row("SELECT COUNT(*) FROM files", [], |r| r.get(0))
        .unwrap();
    assert_eq!(n, 1);
}

#[test]
fn wildcard_name_ranks_through_the_same_tiers() {
    let (_dir, p) = Scratch::db("wildranks");
    let mut s = Seeder::new(&p, true);
    // `report*` anchors the whole name, so these are tiers 1 and 2 …
    let whole_cs = s.add("report2024.pdf", "/a", 1, None);
    let whole_ci = s.add("Report2024.pdf", "/b", 2, None);
    // … and a name that merely contains the pattern is tier 3/4; a trailing
    // star can match nothing, so "report" mid-name counts too.
    let sub_cs = s.add("my-report-final.txt", "/c", 3, None);
    let sub_ci = s.add("my-Report-final.txt", "/d", 4, None);
    let suffix = s.add("2024report.pdf", "/e", 5, None);
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "report*", &SearchOptions::default());
    assert_eq!(
        hits.iter()
            .map(|h| (h.file_id, h.stage))
            .collect::<Vec<_>>(),
        // Within rank 3 the tie breaks by name: "2024…" sorts first.
        vec![
            (whole_cs, 1),
            (whole_ci, 2),
            (suffix, 3),
            (sub_cs, 3),
            (sub_ci, 4)
        ],
        "wildcard terms rank exactly like literal ones"
    );

    let (ordered, _) = run_collect(&conn, "report*2024", &SearchOptions::default());
    assert!(
        !ordered.iter().any(|h| h.name == "2024report.pdf"),
        "segments must match in order"
    );
}

#[test]
fn extension_glob_whole_matches_every_such_file() {
    let (_dir, p) = Scratch::db("extglob");
    let mut s = Seeder::new(&p, true);
    let a = s.add("alpha.txt", "/d", 1, None);
    let b = s.add("beta.txt", "/d", 2, None);
    let upper = s.add("GAMMA.TXT", "/d", 3, None);
    let _other = s.add("delta.pdf", "/d", 4, None);
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "*.txt", &SearchOptions::default());
    let mut tier1: Vec<i64> = hits
        .iter()
        .filter(|h| h.stage == 1)
        .map(|h| h.file_id)
        .collect();
    tier1.sort();
    assert_eq!(tier1, vec![a, b], "every exact-case .txt is a tier-1 hit");
    assert_eq!(
        hits.iter()
            .filter(|h| h.stage == 2)
            .map(|h| h.file_id)
            .collect::<Vec<_>>(),
        vec![upper],
        "case-folded whole match lands at tier 2"
    );
    assert!(!hits.iter().any(|h| h.name == "delta.pdf"));
}

#[test]
fn wildcard_fulltext_narrows_with_fts_and_verifies_order() {
    let (_dir, p) = Scratch::db("wildtext");
    let mut s = Seeder::new(&p, true);
    let ordered = s.add("a.txt", "/d", 1, Some("a wondrous world indeed"));
    // FTS AND-of-segments finds this too (both trigram runs occur), but the
    // pattern requires "wond" before "world" — verification drops it.
    let _reversed = s.add("b.txt", "/d", 2, Some("world of wonders"));
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "wond*world", &SearchOptions::default());
    assert_eq!(
        hits.iter()
            .map(|h| (h.file_id, h.stage))
            .collect::<Vec<_>>(),
        vec![(ordered, 5)],
        "unordered FTS candidates must fail pattern verification"
    );
    let snip = hits[0]
        .snippet
        .as_ref()
        .expect("wildcard hit has a snippet");
    assert_eq!(snip.ranges.len(), 1);
    let (a, b) = snip.ranges[0];
    assert_eq!(&snip.window[a..b], "wondrous world");
}

#[test]
fn wildcard_with_short_segments_falls_back_to_a_full_scan() {
    let (_dir, p) = Scratch::db("wildshort");
    let mut s = Seeder::new(&p, true);
    let hit = s.add("doc.txt", "/d", 1, Some("zz abXcd zz"));
    let _miss = s.add("other.txt", "/d", 2, Some("cd before ab"));
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "ab*cd", &SearchOptions::default());
    assert_eq!(
        hits.iter()
            .map(|h| (h.file_id, h.stage))
            .collect::<Vec<_>>(),
        vec![(hit, 5)]
    );
}

/// The `LIKE` prefilter a straddling wildcard gets must be a *superset* of
/// what the classifier accepts, or real hits vanish silently. Seeds rows
/// exercising every way a segment can sit across `parent || name`, and
/// checks the prefiltered pass against the unfiltered one — obtained via a
/// pattern whose segments all contain a separator, the arm that still scans,
/// so both sides run through the real cascade.
#[test]
fn a_wildcard_prefilter_never_loses_a_hit_the_full_scan_finds() {
    let (_dir, p) = Scratch::db("wildprefilter");
    let mut s = Seeder::new(&p, true);
    let in_name = s.add("report-q3.txt", "/data", 1, None);
    let in_parent = s.add("a.bin", "/reports/q3", 2, None);
    // The pattern's `%` spans the parent/name boundary.
    let across = s.add("q3.txt", "/reports", 3, None);
    let folded = s.add("REPORT-Q3.TXT", "/upper", 4, None);
    let miss = s.add("summary.txt", "/data", 5, None);
    // The trap the separator-free rule exists for: the only way to read this
    // row as a match spans the boundary, and the spanning text contains the
    // separator itself.
    let boundary = s.add("q3.txt", "/x/rep", 6, None);
    // Makes the separator-free rule load-bearing: against `e/pq*txt` the
    // longest segment `e/pq` occurs in `parent || name` only across the
    // join, so anchoring on it would drop this row; the rule falls to `txt`.
    let straddling = s.add("pq.txt", "/a/re", 7, None);
    let conn = s.done();

    let opts = SearchOptions::default();
    for query in ["rep*q3", "rep*rt", "*report*", "re*or*q3", "e/pq*txt"] {
        let (hits, _) = run_collect(&conn, query, &opts);
        let mut got: Vec<i64> = hits.iter().map(|h| h.file_id).collect();
        got.sort();

        // The reference set is computed directly: a row is expected exactly
        // when the compiled pattern matches its name or its full path.
        let split = split_for_cascade(query).unwrap();
        let mut want: Vec<i64> = [
            (in_name, "/data/report-q3.txt"),
            (in_parent, "/reports/q3/a.bin"),
            (across, "/reports/q3.txt"),
            (folded, "/upper/REPORT-Q3.TXT"),
            (miss, "/data/summary.txt"),
            (boundary, "/x/rep/q3.txt"),
            (straddling, "/a/re/pq.txt"),
        ]
        .iter()
        .filter(|(_, path)| {
            let name = path.rsplit('/').next().unwrap();
            split.pattern.find_first(name, true).is_some()
                || split.pattern.find_first(path, true).is_some()
        })
        .map(|(id, _)| *id)
        .collect();
        want.sort();

        assert_eq!(got, want, "query {:?}", query);
    }
}

/// A pattern every segment of which carries a separator has nothing to anchor
/// on, so the pass falls back to scanning — and must still find its hits.
#[test]
fn a_wildcard_with_only_separator_segments_still_scans_and_matches() {
    let (_dir, p) = Scratch::db("wildnoanchor");
    let mut s = Seeder::new(&p, true);
    let hit = s.add("q3.txt", "/a/reports", 1, None);
    let _miss = s.add("q3.txt", "/a/summaries", 2, None);
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "a/rep*rts/", &SearchOptions::default());
    assert_eq!(
        hits.iter().map(|h| h.file_id).collect::<Vec<_>>(),
        vec![hit]
    );
}

#[test]
fn wildcard_path_tier_and_filters() {
    let (_dir, p) = Scratch::db("wildpath");
    let mut s = Seeder::new(&p, true);
    let dir_hit = s.add("a.bin", "/Vacation-2024", 1, None);
    let filtered = s.add("b.bin", "/elsewhere/Vacation-2023", 2, None);
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "Vac*tion", &SearchOptions::default());
    let mut ids: Vec<i64> = hits.iter().map(|h| h.file_id).collect();
    ids.sort();
    let mut want = vec![dir_hit, filtered];
    want.sort();
    assert_eq!(ids, want);
    assert!(hits.iter().all(|h| h.stage == 9), "term only in the path");

    let (kept, _) = run_collect(&conn, "Vac*tion path:/elsewhere", &SearchOptions::default());
    assert_eq!(
        kept.iter().map(|h| h.file_id).collect::<Vec<_>>(),
        vec![filtered]
    );
}

#[test]
fn wildcard_terms_skip_the_fuzzy_stages() {
    let (_dir, p) = Scratch::db("wildfuzzy");
    let mut s = Seeder::new(&p, true);
    let real = s.add("report.txt", "/d", 1, None);
    // A 2-edit typo: fuzzy would admit it for a literal term.
    let _typo = s.add("Reprot.txt", "/d", 2, None);
    let _typo_body = s.add("body.txt", "/d", 3, Some("the reoprt went missing"));
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "rep*rt", &fuzzy_options());
    assert_eq!(
        hits.iter().map(|h| h.file_id).collect::<Vec<_>>(),
        vec![real],
        "no stage 7/8/11 hits for a wildcard term even with fuzzy on"
    );
}

#[test]
fn regex_only_query_hits_name_content_and_path() {
    let (_dir, p) = Scratch::db("regexonly");
    let mut s = Seeder::new(&p, true);
    let by_name = s.add("qz42.txt", "/d", 1, None);
    let by_content = s.add("notes.txt", "/d", 2, Some("ref qz7 in the body"));
    let by_path = s.add("b.bin", "/qz99-dir", 3, None);
    let _miss = s.add("plain.txt", "/d", 4, Some("nothing here"));
    let conn = s.done();

    let (hits, _) = run_collect(&conn, r"regex:qz\d+", &SearchOptions::default());
    assert_eq!(
        hits.iter()
            .map(|h| (h.file_id, h.stage))
            .collect::<Vec<_>>(),
        vec![(by_name, 4), (by_content, 6), (by_path, 10)],
        "regex-only reuses the name/content/path tiers in cascade order"
    );

    let name_snip = hits[0].snippet.as_ref().unwrap();
    let (a, b) = name_snip.ranges[0];
    assert_eq!(&name_snip.window[a..b], "qz42");
    let path_snip = hits[2].snippet.as_ref().unwrap();
    let (a, b) = path_snip.ranges[0];
    assert_eq!(&path_snip.window[a..b], "qz99");
    let body_snip = hits[1].snippet.as_ref().unwrap();
    let (a, b) = body_snip.ranges[0];
    assert_eq!(&body_snip.window[a..b], "qz7");
}

#[test]
fn regex_is_case_insensitive_by_default_and_respects_filters() {
    let (_dir, p) = Scratch::db("regexci");
    let mut s = Seeder::new(&p, true);
    let keep = s.add("QZ1.txt", "/keep", 1, None);
    let _skip = s.add("qz2.txt", "/skip", 2, None);
    let conn = s.done();

    let (hits, _) = run_collect(&conn, r"regex:qz\d path:/keep", &SearchOptions::default());
    assert_eq!(
        hits.iter().map(|h| h.file_id).collect::<Vec<_>>(),
        vec![keep]
    );

    let (cs, _) = run_collect(&conn, r"regex:(?-i:qz)\d", &SearchOptions::default());
    assert!(!cs.iter().any(|h| h.file_id == keep));
}

#[test]
fn regex_alongside_a_term_is_an_accept_predicate() {
    let (_dir, p) = Scratch::db("regexpred");
    let mut s = Seeder::new(&p, true);
    // Kept via the lazy content fetch — the regex is nowhere in its name/path.
    let kept = s.add("budget-a.txt", "/d", 1, Some("code acme7 inside"));
    let _dropped = s.add("budget-b.txt", "/d", 2, None);
    let conn = s.done();

    let (hits, _) = run_collect(&conn, r"budget regex:acme\d", &SearchOptions::default());
    assert_eq!(
        hits.iter()
            .map(|h| (h.file_id, h.stage))
            .collect::<Vec<_>>(),
        vec![(kept, 3)],
        "the term drives ranking; the regex gates acceptance"
    );
}

#[test]
fn hostile_regexes_complete_quickly() {
    let (_dir, p) = Scratch::db("regexhostile");
    let mut s = Seeder::new(&p, true);
    s.add("aaa.txt", "/d", 1, Some(&"a".repeat(50_000)));
    let conn = s.done();

    let start = std::time::Instant::now();
    let (hits, _) = run_collect(&conn, r#"regex:"(a+)+$""#, &SearchOptions::default());
    assert!(!hits.is_empty(), "the all-a body does end in a run of a's");
    assert!(
        start.elapsed() < std::time::Duration::from_secs(5),
        "hostile regex must not blow up: took {:?}",
        start.elapsed()
    );
}

#[test]
fn service_surfaces_invalid_regex_as_an_error() {
    let (_dir, p) = Scratch::db("regexerr");
    let mut s = Seeder::new(&p, true);
    s.add("anything.txt", "/d", 1, None);
    drop(s.done());

    let (service, updates) = SearchService::new(p.clone(), Arc::new(|| {}));
    let generation = service.search("regex:[", SearchOptions::default());
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
    let mut message = None;
    while std::time::Instant::now() < deadline {
        match updates.recv_timeout(std::time::Duration::from_millis(200)) {
            Ok(SearchUpdate::Error {
                generation: g,
                message: m,
            }) if g == generation => {
                message = Some(m);
                break;
            }
            Ok(_) => {}
            Err(_) => break,
        }
    }
    let message = message.expect("invalid regex must surface as a search error");
    assert!(message.contains("regex"), "unhelpful message: {}", message);
    service.shutdown();
}

#[test]
fn generation_bump_cancels_mid_stream() {
    let (_dir, p) = Scratch::db("cancel");
    let mut s = Seeder::new(&p, true);
    for i in 0..500 {
        s.add(&format!("bulk-{:04}.txt", i), "/d", 1, None);
    }
    let conn = s.done();

    let split = split_for_cascade("bulk").unwrap();
    let latest = Arc::new(AtomicU64::new(3));
    let latest_for_sink = latest.clone();
    let mut batches = 0usize;
    let outcome = cascade::run(
        &conn,
        &split,
        &SearchOptions::default(),
        3,
        &latest,
        &mut |_batch| {
            batches += 1;
            // Simulate a new keystroke arriving after the first batch.
            latest_for_sink.store(99, Ordering::SeqCst);
        },
    )
    .unwrap();
    assert!(outcome.is_none(), "cancelled search must not complete");
    assert_eq!(batches, 1, "no further batches after the generation moved");
}

#[test]
fn service_rapid_fire_completes_only_the_last_generation() {
    let (_dir, p) = Scratch::db("service");
    let mut s = Seeder::new(&p, true);
    for i in 0..2000 {
        s.add(
            &format!("file-{:04}.txt", i),
            "/d",
            1,
            Some("shared corpus body text"),
        );
    }
    drop(s.done());

    let (service, updates) = SearchService::new(p.clone(), Arc::new(|| {}));
    for i in 0..50 {
        service.search(&format!("corpus body {}", i % 3), SearchOptions::default());
    }
    // Final query that actually matches, so completion carries hits too.
    let last_gen = service.search("corpus", SearchOptions::default());

    let mut completed: Vec<u64> = Vec::new();
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
    while std::time::Instant::now() < deadline {
        match updates.recv_timeout(std::time::Duration::from_millis(200)) {
            Ok(SearchUpdate::Completed { generation, .. }) => {
                completed.push(generation);
                if generation == last_gen {
                    break;
                }
            }
            Ok(_) => {}
            Err(std::sync::mpsc::RecvTimeoutError::Timeout) => continue,
            Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
        }
    }
    assert_eq!(
        completed.last().copied(),
        Some(last_gen),
        "the newest generation must be the one that completes (saw {:?})",
        completed
    );
    service.shutdown();
}

#[test]
fn service_reports_missing_db_as_error() {
    let (_dir, missing) = Scratch::db("missing");
    let (service, updates) = SearchService::new(missing.clone(), Arc::new(|| {}));
    let generation = service.search("anything", SearchOptions::default());
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
    let mut got_error = false;
    while std::time::Instant::now() < deadline {
        match updates.recv_timeout(std::time::Duration::from_millis(200)) {
            Ok(SearchUpdate::Error { generation: g, .. }) if g == generation => {
                got_error = true;
                break;
            }
            Ok(_) => {}
            Err(_) => break,
        }
    }
    assert!(got_error, "missing index must surface as a search error");
    service.shutdown();
}

/// Collect the names one search returns, or the error it produced. The
/// worker keeps its connection between requests, so these tests run several
/// searches against one service — the situation the reuse has to survive.
fn search_names(
    service: &SearchService,
    updates: &std::sync::mpsc::Receiver<SearchUpdate>,
    query: &str,
) -> Result<Vec<String>, String> {
    let generation = service.search(query, SearchOptions::default());
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
    let mut names = Vec::new();
    while std::time::Instant::now() < deadline {
        match updates.recv_timeout(std::time::Duration::from_millis(200)) {
            Ok(SearchUpdate::Hits {
                generation: g,
                hits,
            }) if g == generation => {
                names.extend(hits.into_iter().map(|h| h.name));
            }
            Ok(SearchUpdate::Completed { generation: g, .. }) if g == generation => {
                names.sort();
                return Ok(names);
            }
            Ok(SearchUpdate::Error {
                generation: g,
                message,
            }) if g == generation => return Err(message),
            Ok(_) => {}
            Err(_) => break,
        }
    }
    panic!("search {:?} never completed", query);
}

#[test]
fn repeated_searches_on_one_service_stay_correct() {
    let (_dir, p) = Scratch::db("warmrepeat");
    let mut s = Seeder::new(&p, true);
    s.add("alpha.txt", "/d", 1, Some("alpha body"));
    s.add("beta.txt", "/d", 1, Some("beta body"));
    drop(s.done());

    let (service, updates) = SearchService::new(p.clone(), Arc::new(|| {}));
    for _ in 0..3 {
        assert_eq!(
            search_names(&service, &updates, "alpha").unwrap(),
            vec!["alpha.txt"]
        );
        assert_eq!(
            search_names(&service, &updates, "beta").unwrap(),
            vec!["beta.txt"]
        );
    }
    service.shutdown();
}

/// A rebuild replaces the file at the *same path*, so nothing about the path
/// tells a held connection it now looks at a deleted inode — the reason
/// `db::index_epoch` exists.
#[test]
fn a_rebuilt_index_at_the_same_path_is_picked_up() {
    let (_dir, p) = Scratch::db("warmrebuild");
    let mut s = Seeder::new(&p, true);
    s.add("before.txt", "/d", 1, Some("shared body"));
    drop(s.done());

    let (service, updates) = SearchService::new(p.clone(), Arc::new(|| {}));
    assert_eq!(
        search_names(&service, &updates, "shared").unwrap(),
        vec!["before.txt"],
        "the pre-rebuild index answers first"
    );

    // A rebuild, spelled the way the coordinator spells one: the wipe branch
    // of `open_or_recreate` is the real call site that bumps the epoch, so
    // this exercises the actual invalidation rather than poking the counter.
    for suffix in ["", "-wal", "-shm"] {
        std::fs::remove_file(format!("{}{}", p.display(), suffix)).ok();
    }
    let mut s = Seeder::new(&p, true);
    s.add("after.txt", "/d", 1, Some("shared body"));
    drop(s.done());

    assert_eq!(
        search_names(&service, &updates, "shared").unwrap(),
        vec!["after.txt"],
        "a held connection served the replaced index"
    );
    service.shutdown();
}

#[test]
fn set_db_path_repoints_a_held_connection() {
    let (_dir1, first) = Scratch::db("warmpath1");
    let mut s = Seeder::new(&first, true);
    s.add("infirst.txt", "/d", 1, Some("shared body"));
    drop(s.done());

    let (_dir2, second) = Scratch::db("warmpath2");
    let mut s = Seeder::new(&second, true);
    s.add("insecond.txt", "/d", 1, Some("shared body"));
    drop(s.done());

    let (service, updates) = SearchService::new(first.clone(), Arc::new(|| {}));
    assert_eq!(
        search_names(&service, &updates, "shared").unwrap(),
        vec!["infirst.txt"]
    );

    service.set_db_path(second.clone());
    assert_eq!(
        search_names(&service, &updates, "shared").unwrap(),
        vec!["insecond.txt"],
        "the connection stayed on the old index after the path moved"
    );
    service.shutdown();
}

#[test]
fn a_failed_search_does_not_wedge_the_next_one() {
    let (_dir, p) = Scratch::db("warmrecover");
    let (service, updates) = SearchService::new(p.clone(), Arc::new(|| {}));

    search_names(&service, &updates, "anything").expect_err("no index yet");

    let mut s = Seeder::new(&p, true);
    s.add("arrived.txt", "/d", 1, Some("shared body"));
    drop(s.done());

    assert_eq!(
        search_names(&service, &updates, "shared").unwrap(),
        vec!["arrived.txt"],
        "the failure left the worker unable to open the index that appeared"
    );
    service.shutdown();
}

/// The connection is released once a search session goes quiet: a deleted
/// index keeps its blocks until the last handle closes, and an open reader
/// stops SQLite from resetting the WAL. Checked through `/proc/self/fd`;
/// a reopen-still-works assertion would pass either way.
#[cfg(target_os = "linux")]
#[test]
fn the_connection_is_released_once_searching_stops() {
    let (_dir, p) = Scratch::db("warmrelease");
    let mut s = Seeder::new(&p, true);
    s.add("held.txt", "/d", 1, Some("shared body"));
    drop(s.done());

    let holds_index = || {
        let Ok(entries) = std::fs::read_dir("/proc/self/fd") else {
            return false;
        };
        entries
            .filter_map(|e| e.ok())
            .filter_map(|e| std::fs::read_link(e.path()).ok())
            .any(|target| target == p)
    };

    let (service, updates) = SearchService::new_with_idle_release(
        p.clone(),
        Arc::new(|| {}),
        std::time::Duration::from_millis(150),
    );
    assert_eq!(
        search_names(&service, &updates, "shared").unwrap(),
        vec!["held.txt"]
    );
    assert!(
        holds_index(),
        "the connection should be held across requests"
    );

    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
    while holds_index() && std::time::Instant::now() < deadline {
        std::thread::sleep(std::time::Duration::from_millis(50));
    }
    assert!(!holds_index(), "the idle connection was never released");

    assert_eq!(
        search_names(&service, &updates, "shared").unwrap(),
        vec!["held.txt"]
    );
    service.shutdown();
}

/// An explicit release must leave the file closed even when a search was
/// queued a moment before it: `release_connection` bumps the generation
/// first, so the queued request is stale and the worker drops it instead of
/// reopening the index while the caller is on its way to delete it.
///
/// The generation is checked rather than the descriptor because the race is
/// won or lost in microseconds; an fd assertion would pass with the bug in
/// place most of the time.
#[test]
fn a_release_supersedes_a_search_queued_before_it() {
    let (_dir, p) = Scratch::db("relcancel");
    let mut s = Seeder::new(&p, true);
    s.add("held.txt", "/d", 1, Some("shared body"));
    drop(s.done());

    let (service, updates) = SearchService::new_with_idle_release(
        p.clone(),
        Arc::new(|| {}),
        std::time::Duration::from_secs(60),
    );

    let queued = service.search("shared", SearchOptions::default());
    service.release_connection();
    let after = service.search("shared", SearchOptions::default());
    assert!(
        after > queued + 1,
        "the release must advance the generation past {}, got {}",
        queued,
        after
    );

    assert_eq!(
        search_names(&service, &updates, "shared").unwrap(),
        vec!["held.txt"]
    );
    service.shutdown();
}

/// A pass hands hits over *while* it scans. Proven by ordering rather than
/// by batch count — `flush_pass` has always chunked its output, so counting
/// sink calls proves nothing. Pass A scans in stored `(parent, name)` order,
/// so a *worse* match seeded early must arrive before a *better* one at a
/// late path; emitting at the end would sort them and lead with rank 1.
#[test]
fn a_pass_hands_hits_over_before_the_scan_reaches_the_end() {
    let (_dir, p) = Scratch::db("stream");
    let mut s = Seeder::new(&p, true);
    let early_worse = s.add("my_zebra_file.txt", "/aaa", 1, None); // rank 3
    for i in 0..400 {
        s.add(&format!("filler{:04}.txt", i), "/mmm", i as u64 + 2, None);
    }
    let content = s.add("unrelated.txt", "/nnn", 998, Some("a zebra in the text")); // rank 5
    let late_better = s.add("zebra", "/zzz", 999, None); // rank 1
    let conn = s.done();

    let options = SearchOptions {
        batch: 100,
        ..SearchOptions::default()
    };
    let (batches, outcome) = run_collect_batches(&conn, "zebra", &options);

    assert_eq!(outcome.total, 3, "every match still reaches the sink");
    assert_eq!(
        batches
            .first()
            .map(|b| b.as_slice())
            .and_then(|b| b.first())
            .map(|h| h.file_id),
        Some(early_worse),
        "the early hit should have gone out before the scan found the better one; \
         batch sizes: {:?}",
        batches.iter().map(|b| b.len()).collect::<Vec<_>>()
    );

    // ...and sorting the stream the way the GUI does still puts rank 1 on
    // top: streaming must not change *what* a search returns, only when.
    for batch in [1usize, 2, 100] {
        let options = SearchOptions {
            batch,
            ..SearchOptions::default()
        };
        let (hits, outcome) = run_collect(&conn, "zebra", &options);
        assert_eq!(
            hits.iter().map(|h| h.file_id).collect::<Vec<_>>(),
            vec![late_better, early_worse, content],
            "batch size {} must not change the set or its ranking",
            batch
        );
        assert_eq!(outcome.total, 3, "batch size {}", batch);
    }
}

/// The time bound, which a size-only rule would miss: a query matching a
/// handful of rows out of many still paints them as the scan reaches them
/// rather than at the end.
#[test]
fn a_sparse_match_still_streams_before_the_scan_ends() {
    let (_dir, p) = Scratch::db("sparse");
    let mut s = Seeder::new(&p, true);
    for i in 0..3000 {
        let name = if i % 1000 == 500 {
            format!("needle{:04}.txt", i)
        } else {
            format!("hay{:04}.txt", i)
        };
        s.add(&name, "/d", i as u64 + 1, Some("body"));
    }
    let conn = s.done();

    let options = SearchOptions {
        batch: 100,
        ..SearchOptions::default()
    };
    let (batches, outcome) = run_collect_batches(&conn, "needle", &options);

    assert_eq!(outcome.total, 3, "all three needles found");
    assert_eq!(batches.iter().map(|b| b.len()).sum::<usize>(), 3);
    assert!(
        !batches.is_empty() && batches[0].len() < 3,
        "the first needle should not have waited for the other two; batches: {:?}",
        batches.iter().map(|b| b.len()).collect::<Vec<_>>()
    );
}

// ---------------------------------------------------------------------------
// Fuzzy-snippet regressions, three symptoms of one shape: bitap reports
// only where a match *ends*, and the mark's start and extent are derived
// from that — the whole-field window, the start (`end - term.len()`), and
// the stop-at-first-accept extent each went wrong. One test per symptom.
// ---------------------------------------------------------------------------

/// The window of a fuzzy name/path hit is the whole field, never a suffix of
/// it, so the ranges index the field the frontend paints.
#[test]
fn fuzzy_name_and_path_snippets_carry_the_whole_field() {
    let (_dir, p) = Scratch::db("fuzzy-whole-field");
    let mut s = Seeder::new(&p, true);
    // The match sits in the last third of the name, which is what used to
    // push the window's start off zero.
    s.add(
        "a_long_and_deliberately_padded_out_quarterly_repot.txt",
        "/home/me/documents/archive",
        1,
        None,
    );
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "report", &fuzzy_options());
    let hit = hits
        .iter()
        .find(|h| h.stage == 7)
        .expect("a fuzzy filename hit");

    let snip = hit.snippet.as_ref().expect("a name hit carries a snippet");
    assert_eq!(snip.window, hit.name, "the window is not the whole name");
    assert!(!snip.truncated_start, "the name was windowed");
    assert!(!snip.truncated_end, "the name was windowed");
    for &(a, b) in &snip.ranges {
        assert!(b <= hit.name.len(), "range {a}..{b} runs past the name");
        assert!(
            hit.name.is_char_boundary(a) && hit.name.is_char_boundary(b),
            "range {a}..{b} is not on char boundaries"
        );
    }
}

/// The mark covers the matched text and nothing else — `repot` used to mark
/// `1Repo` inside `1Reporter`, reaching back over the digit.
#[test]
fn a_fuzzy_mark_covers_the_matched_text_and_nothing_else() {
    let (_dir, p) = Scratch::db("fuzzy-mark-span");
    let mut s = Seeder::new(&p, true);
    // The leading digit is the point: it is what the old range reached back
    // over. In the body too, for the full-text tier.
    s.add("1Reporter.txt", "/home/me/docs", 1, None);
    s.add(
        "body.txt",
        "/home/me/docs",
        2,
        Some("filed under 1Reporter last week"),
    );
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "repot", &fuzzy_options());

    // Stage 7 — fuzzy filename, marked inside the whole name.
    let name_hit = hits
        .iter()
        .find(|h| h.stage == 7)
        .expect("a fuzzy filename hit");
    let snip = name_hit.snippet.as_ref().expect("name tiers carry one");
    let (a, b) = snip.ranges[0];
    assert_eq!(
        &snip.window[a..b],
        "Repo",
        "the mark is {:?}; it must cover the match and not the leading digit",
        &snip.window[a..b]
    );

    // Stage 8 — fuzzy full text, marked inside the snippet window.
    let body_hit = hits
        .iter()
        .find(|h| h.stage == 8)
        .expect("a fuzzy full-text hit");
    let snip = body_hit.snippet.as_ref().expect("content tiers carry one");
    let (a, b) = snip.ranges[0];
    assert_eq!(
        &snip.window[a..b],
        "Repo",
        "the mark is {:?}; it must cover the match and not the leading digit",
        &snip.window[a..b]
    );
}

/// The mark extends to the best alignment rather than stopping at the first
/// accept: `quarterly` over `quartrly` must light the whole word (one edit),
/// not the two-edit `quartrl` prefix.
#[test]
fn a_fuzzy_mark_is_not_truncated_to_a_leading_part_of_the_term() {
    let (_dir, p) = Scratch::db("fuzzy-mark-full");
    let mut s = Seeder::new(&p, true);
    // The name must not match at all: a row the filename tier claims never
    // reaches the full-text tier. And the body must hold a fuzzy *variant* —
    // the term verbatim would be an exact content match, stage 5 or 6.
    s.add(
        "notes.txt",
        "/home/me/docs",
        1,
        Some("the quartrly budget was revised"),
    );
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "quarterly", &fuzzy_options());
    let hit = hits
        .iter()
        .find(|h| h.stage == 8)
        .expect("a fuzzy full-text hit");
    let snip = hit.snippet.as_ref().expect("content tiers carry one");
    let (a, b) = snip.ranges[0];
    assert_eq!(
        &snip.window[a..b],
        "quartrly",
        "the mark is {:?}, a leading part of what matched",
        &snip.window[a..b]
    );
}

/// `files` stores `parent` and `name` with no concatenation, so a one-piece
/// term uses `name LIKE ? OR parent LIKE ?` — equivalent, since a
/// separator-free term cannot span the boundary. A multi-segment wildcard
/// *can* span it and neither half fires; those fall back to scanning, and
/// this fixture fails if that fallback is ever "optimised" away.
#[test]
fn a_wildcard_spanning_the_directory_boundary_is_still_found() {
    let (_dir, p) = Scratch::db("pathstraddle");
    let mut s = Seeder::new(&p, true);
    let straddling = s.add("q3.txt", "/x/docs", 1, None);
    // Same two pieces, both inside the name: found either way, so it proves
    // the query ran rather than that the fallback was reached.
    let in_name = s.add("doc-q3.txt", "/other", 2, None);
    let _miss = s.add("q3.txt", "/x/plans", 3, None);
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "doc*q3", &SearchOptions::default());
    let mut ids: Vec<i64> = hits.iter().map(|h| h.file_id).collect();
    ids.sort();
    let mut want = vec![straddling, in_name];
    want.sort();
    assert_eq!(ids, want, "the straddling path match must not be dropped");

    let hit = hits.iter().find(|h| h.file_id == straddling).unwrap();
    assert_eq!(hit.stage, 9);
    assert_eq!(hit.path, "/x/docs/q3.txt", "parent and name rejoined");
}

/// `folder:` must reach the folder's own files, not just its
/// subdirectories': the single `parent LIKE 'dir/%'` covers both only
/// because every stored parent ends in a separator.
#[test]
fn the_folder_filter_covers_the_folder_itself_and_its_subtree() {
    let (_dir, p) = Scratch::db("folderself");
    let mut s = Seeder::new(&p, true);
    let own = s.add("top.txt", "/srv/data", 1, None);
    let nested = s.add("deep.txt", "/srv/data/2024", 2, None);
    let _sibling = s.add("other.txt", "/srv/data-archive", 3, None);
    let _outside = s.add("far.txt", "/srv", 4, None);
    let conn = s.done();

    let (hits, _) = run_collect(&conn, "txt folder:/srv/data", &SearchOptions::default());
    let mut ids: Vec<i64> = hits.iter().map(|h| h.file_id).collect();
    ids.sort();
    let mut want = vec![own, nested];
    want.sort();
    assert_eq!(ids, want, "the folder's own files count as inside it");
}