// WRITE_LOCK guard is held across .await inside block_on().
// This is safe — block_on runs single-threaded, no concurrent tasks can deadlock.
#![allow(clippy::await_holding_lock)]
//! Sparse vector storage for SPLADE hybrid search.
//!
//! ## Invalidation contract (v19+)
//!
//! The persisted `splade.index.bin` file embeds a generation counter in its
//! header. On load, the counter is compared against `metadata.splade_generation`;
//! any mismatch forces a rebuild from SQLite. The counter is maintained by
//! [`Store::bump_splade_generation_tx`], the single source of truth for this
//! bump. **Every write site that mutates `sparse_vectors` must call it before
//! committing.** The `sparse_vectors` table has a foreign key with
//! `ON DELETE CASCADE` on `chunk_id → chunks(id)`, so deletes through the
//! `chunks` table automatically cascade — those paths bump the generation via
//! their own `bump_splade_generation_tx` call after the `chunks` delete.
//!
//! The counter additionally serves as the store-wide *write generation* for
//! the HNSW sidecar stamp (`hnsw::StoreStamp`): dense-embedding rewrites
//! (`update_embeddings_with_hashes_batch`, the enrichment pass) bump it too,
//! even though they never touch `sparse_vectors` — the dirty-flag self-heal
//! relies on the counter moving for every vector-affecting write. The
//! resulting spurious SPLADE rebuild after an enrichment pass is the
//! accepted cost of keeping one counter.
//!
//! Do not add another vector-affecting write site without wiring the bump.

use super::{ReadWrite, Store};
use crate::splade::SparseVector;
use crate::store::StoreError;

use sqlx::Row;

/// Default number of distinct chunk_ids fetched per page by
/// [`Store::load_all_sparse_vectors`].
///
/// Sized so the per-batch row Vec fits comfortably in a few MB even at the
/// upper bound of tokens-per-chunk (~100-500). With 1000 chunks × 200 avg
/// tokens/row × ~80B/row ≈ 16 MB per batch. An unpaginated single-query
/// `fetch_all` could hit multiple GB on a large index. Larger values reduce
/// round-trip count but cap the memory win.
const LOAD_SPARSE_CHUNK_ID_BATCH_DEFAULT: i64 = 1000;

/// Resolve the SPLADE load page size honoring `CQS_LOAD_SPARSE_BATCH`.
/// Falls back to [`LOAD_SPARSE_CHUNK_ID_BATCH_DEFAULT`] when unset, empty,
/// unparseable, or zero. Resolved as a `u64` (the page size is always
/// positive) and narrowed to the `i64` the SQL `LIMIT` bind expects;
/// `i64::MAX`-and-up clamps to `i64::MAX` so a pathological env value can't
/// wrap negative.
fn load_sparse_chunk_id_batch() -> i64 {
    let n = crate::limits::parse_env_u64(
        "CQS_LOAD_SPARSE_BATCH",
        LOAD_SPARSE_CHUNK_ID_BATCH_DEFAULT as u64,
    );
    n.min(i64::MAX as u64) as i64
}

/// Bump the SPLADE generation counter inside an existing write transaction.
///
/// This is the one place the counter gets incremented. Every write path that
/// touches `sparse_vectors` (directly or via FK cascade from `chunks`) must
/// call this before committing. Silent parse failures on corrupt metadata
/// are warned but not made fatal — a corrupt value resets to 1 and the next
/// loader sees a fresh generation, forcing a clean rebuild.
///
/// Takes `&mut Transaction` so callers that already hold a write transaction
/// don't fight the WRITE_LOCK. Readers use [`Store::splade_generation`]
/// against the pool.
pub(crate) async fn bump_splade_generation_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
) -> Result<(), StoreError> {
    let current: Option<(String,)> =
        sqlx::query_as("SELECT value FROM metadata WHERE key = 'splade_generation'")
            .fetch_optional(&mut **tx)
            .await?;
    let next: u64 = match current {
        Some((ref s,)) => match s.parse::<u64>() {
            Ok(n) => n.saturating_add(1),
            Err(e) => {
                tracing::warn!(
                    raw = %s,
                    error = %e,
                    "splade_generation metadata is not a valid u64, resetting to 1",
                );
                1
            }
        },
        None => 1,
    };
    sqlx::query(
        "INSERT INTO metadata (key, value) VALUES ('splade_generation', ?1) \
         ON CONFLICT(key) DO UPDATE SET value = excluded.value",
    )
    .bind(next.to_string())
    .execute(&mut **tx)
    .await?;
    Ok(())
}

/// DS2-9: Standalone (not-in-transaction) SPLADE generation bump.
///
/// Used by error paths that already rolled back the primary write transaction
/// but still want to force on-disk `splade.index.bin` invalidation. The upsert
/// attempt may have touched DDL state visible to concurrent writers (DROP /
/// CREATE INDEX), and any cached on-disk SPLADE index built from a prior
/// successful state should be treated as suspect. Bumping in its own micro-
/// transaction is the cheapest belt-and-suspenders guard: the next loader
/// sees the new generation, fails the checksum equality against the file
/// header, and rebuilds from whatever sparse_vectors actually persisted.
///
/// Opens its own write transaction on the pool. Errors here are logged at
/// warn level and returned so callers can surface a combined failure, but
/// the typical pattern is log-and-swallow — the original upsert error is
/// what the user cares about, the bump is a durability safety net.
pub(crate) async fn bump_splade_generation_standalone(
    pool: &sqlx::Pool<sqlx::Sqlite>,
) -> Result<(), StoreError> {
    let mut tx = pool.begin().await?;
    bump_splade_generation_tx(&mut tx).await?;
    tx.commit().await?;
    Ok(())
}

impl Store<ReadWrite> {
    /// Upsert sparse vectors for a batch of chunks.
    /// Replaces existing vectors for the same chunk_id.
    ///
    /// **Bulk insert optimization:** drops the secondary `idx_sparse_token`
    /// index before the bulk insert and recreates it after. With SPLADE-Code
    /// 0.6B's denser sparse vectors (~1000+ tokens per chunk), the secondary
    /// B-tree maintenance per row was the dominant cost — every INSERT had to
    /// update both the PRIMARY KEY index AND the token_id index, doubling
    /// per-row work and producing 6+ GB WAL files for a single reindex pass.
    /// Drop + recreate is the standard SQLite bulk-load pattern: the recreate
    /// rebuilds the index in one O(n log n) batch instead of N O(log n) per-row
    /// updates, and the read-time SPLADE search path is unaffected because
    /// the index is back in place by the time the function returns.
    ///
    /// **Lock-hold contract:** the bulk DELETE+INSERT loop runs in
    /// chunked sub-transactions of `CHUNKS_PER_TX` chunks each, releasing
    /// `WRITE_LOCK` (and SQLite's WAL writer lock) between batches. Without
    /// the chunking pass, a single 5614-chunk reindex held the in-process
    /// write lock for ~16 s (8 s DELETE + 8 s CREATE INDEX rebuild),
    /// stalling concurrent daemon RPCs (notes add, audit-mode toggle,
    /// `cqs serve` cluster queries). Now each sub-tx covers ~50 chunks ≈
    /// 50k sparse rows, committing in ~250 ms; the daemon interleaves RPC
    /// work between batches. The CREATE INDEX rebuild remains in its own
    /// short tx at the end (still ~8 s on 1M rows but now bracketed by
    /// short txs), so a daemon RPC arriving mid-rebuild waits at most one
    /// CREATE INDEX call instead of the full 16 s window.
    ///
    /// **Crash safety:** every sub-tx bumps `splade_generation`, so a
    /// crash mid-bulk forces splade.index.bin rebuild on next load
    /// against whatever sparse_vectors actually persisted. Re-running the
    /// upsert with the same `(chunk_id, vec)` pairs is idempotent: each
    /// chunk's old rows get DELETE-then-INSERT'd identically. The
    /// secondary index is `CREATE INDEX IF NOT EXISTS`, so the recreate
    /// pass is also a no-op-on-retry.
    pub fn upsert_sparse_vectors(
        &self,
        vectors: &[(String, SparseVector)],
    ) -> Result<usize, StoreError> {
        let _span = tracing::info_span!("upsert_sparse_vectors", count = vectors.len()).entered();
        if vectors.is_empty() {
            return Ok(0);
        }
        self.rt.block_on(async {
            use crate::store::helpers::sql::max_rows_per_statement;
            // Each row binds 3 vars (chunk_id, token_id, weight). With
            // ~1000 tokens/chunk for SPLADE-Code 0.6B, ROWS_PER_INSERT
            // accommodates ~10 chunks worth of rows per INSERT statement.
            const ROWS_PER_INSERT: usize = max_rows_per_statement(3);
            // CHUNKS_PER_TX caps how long each sub-transaction holds the
            // write lock. ~50 chunks × ~1000 tokens ≈ 50k rows ≈ ~250 ms
            // commit on WAL throughput, which is the right tradeoff
            // between BEGIN/COMMIT round-trips (overhead per tx) and
            // lock-hold time (latency for concurrent writers). Operators
            // can tune via `CQS_SPARSE_CHUNKS_PER_TX` if their workload
            // (e.g. very dense sparse vectors, slower disks) wants
            // smaller batches.
            let chunks_per_tx: usize = std::env::var("CQS_SPARSE_CHUNKS_PER_TX")
                .ok()
                .and_then(|v| v.parse().ok())
                .filter(|n: &usize| *n > 0)
                .unwrap_or(50);

            // Step 1: drop the secondary token_id index in its own short
            // transaction. The PRIMARY KEY index (chunk_id, token_id) is
            // unavoidable; the secondary is purely for read-time SPLADE
            // search lookups and can be safely dropped+recreated around
            // any write batch. We bump `splade_generation` here too so a
            // crash anywhere after this tx commits forces splade.index.bin
            // rebuild on next load — even if the bulk loop never starts,
            // the on-disk index file is invalidated.
            {
                let (_guard, mut tx) = self.begin_write().await?;
                tracing::debug!("Dropping idx_sparse_token before bulk insert");
                sqlx::query("DROP INDEX IF EXISTS idx_sparse_token")
                    .execute(&mut *tx)
                    .await?;
                bump_splade_generation_tx(&mut tx).await?;
                tx.commit().await?;
            }

            // Step 2: chunked DELETE+INSERT loop. Each iteration takes a
            // fresh `begin_write()` so `WRITE_LOCK` is released between
            // batches, letting other in-process writers (daemon RPC
            // threads) interleave. SQLite-side WAL writer lock is also
            // released on `tx.commit()`.
            let inner_result: Result<usize, StoreError> = async {
                let mut total = 0usize;
                for chunk_batch in vectors.chunks(chunks_per_tx) {
                    let (_guard, mut tx) = self.begin_write().await?;

                    // DELETE chunk_ids in this batch. Bound by
                    // `chunks_per_tx`, well under the SQLite variable
                    // limit (32766) so the batch fits one statement.
                    let mut qb: sqlx::QueryBuilder<sqlx::Sqlite> =
                        sqlx::QueryBuilder::new("DELETE FROM sparse_vectors WHERE chunk_id IN (");
                    let mut sep = qb.separated(", ");
                    for (id, _) in chunk_batch {
                        sep.push_bind(id.as_str());
                    }
                    sep.push_unseparated(")");
                    qb.build().execute(&mut *tx).await?;

                    // INSERT rows for chunks in this batch, packed across
                    // chunks so each statement fills close to capacity.
                    let mut pending: Vec<(&str, u32, f32)> = Vec::with_capacity(ROWS_PER_INSERT);
                    for (chunk_id, sparse) in chunk_batch {
                        for &(token_id, weight) in sparse {
                            pending.push((chunk_id.as_str(), token_id, weight));
                            if pending.len() >= ROWS_PER_INSERT {
                                let mut qb: sqlx::QueryBuilder<sqlx::Sqlite> =
                                    sqlx::QueryBuilder::new(
                                        "INSERT INTO sparse_vectors (chunk_id, token_id, weight)",
                                    );
                                qb.push_values(pending.iter(), |mut b, &(cid, tid, w)| {
                                    b.push_bind(cid).push_bind(tid as i64).push_bind(w);
                                });
                                qb.build().execute(&mut *tx).await?;
                                total += pending.len();
                                pending.clear();
                            }
                        }
                    }
                    if !pending.is_empty() {
                        let mut qb: sqlx::QueryBuilder<sqlx::Sqlite> = sqlx::QueryBuilder::new(
                            "INSERT INTO sparse_vectors (chunk_id, token_id, weight)",
                        );
                        qb.push_values(pending.iter(), |mut b, &(cid, tid, w)| {
                            b.push_bind(cid).push_bind(tid as i64).push_bind(w);
                        });
                        qb.build().execute(&mut *tx).await?;
                        total += pending.len();
                    }

                    // Bump generation in EVERY sub-tx so a partial bulk
                    // forces index rebuild on next load. Cheap (one
                    // SELECT + one UPSERT on metadata).
                    bump_splade_generation_tx(&mut tx).await?;
                    tx.commit().await?;
                }
                Ok(total)
            }
            .await;

            // Step 3: recreate the secondary index in its own tx. SQLite
            // builds the index in one O(n log n) pass from the now-fully-
            // populated table — far cheaper than maintaining the index
            // per-row through the bulk insert. Even on partial-bulk
            // failure we attempt the recreate so subsequent reads keep
            // their indexed lookup path.
            match inner_result {
                Ok(total) => {
                    let recreate_result: Result<(), StoreError> = async {
                        let (_guard, mut tx) = self.begin_write().await?;
                        tracing::debug!("Recreating idx_sparse_token after bulk insert");
                        sqlx::query(
                            "CREATE INDEX IF NOT EXISTS idx_sparse_token ON sparse_vectors(token_id)",
                        )
                        .execute(&mut *tx)
                        .await?;
                        bump_splade_generation_tx(&mut tx).await?;
                        tx.commit().await?;
                        Ok(())
                    }
                    .await;
                    if let Err(e) = recreate_result {
                        // The bulk data committed successfully; only the
                        // CREATE INDEX failed. Surface as the call's
                        // overall error so the caller can retry — the
                        // next upsert will attempt the recreate again.
                        tracing::warn!(
                            error = %e,
                            "upsert_sparse_vectors: bulk committed but CREATE INDEX failed; \
                             search path will fall back to full table scan until next upsert"
                        );
                        return Err(e);
                    }
                    tracing::info!(
                        entries = total,
                        chunks = vectors.len(),
                        "Sparse vectors upserted"
                    );
                    Ok(total)
                }
                Err(e) => {
                    // Step 1 already bumped the generation, and every
                    // committed sub-tx in Step 2 also bumped — so any
                    // on-disk splade.index.bin is already invalidated.
                    // Run one more standalone bump as belt-and-suspenders
                    // in case Step 1 itself was the failure point.
                    tracing::warn!(
                        error = %e,
                        "upsert_sparse_vectors failed mid-bulk — bumping SPLADE generation to invalidate on-disk cache"
                    );
                    if let Err(bump_err) = bump_splade_generation_standalone(&self.pool).await {
                        tracing::warn!(
                            error = %bump_err,
                            "failed to bump SPLADE generation on error-path cleanup (non-fatal)"
                        );
                    }
                    Err(e)
                }
            }
        })
    }
}

impl<Mode> Store<Mode> {
    /// Load all sparse vectors for building the in-memory SpladeIndex.
    /// Returns Vec of (chunk_id, sparse_vector).
    ///
    /// Paginates by chunk_id in batches of up to
    /// [`load_sparse_chunk_id_batch`] distinct chunk_ids, using a cursor on
    /// `chunk_id`. Each batch's rows are fully grouped before discarding
    /// the row Vec, so peak memory is bounded to one batch (~5-20 MB) plus
    /// the growing result. A single unpaginated `SELECT ... ORDER BY chunk_id`
    /// + `fetch_all` would instead materialize the entire table — ~500 MB peak
    /// on a 60k-chunk index with ~100 tokens/chunk.
    pub fn load_all_sparse_vectors(&self) -> Result<Vec<(String, SparseVector)>, StoreError> {
        let _span = tracing::info_span!("load_all_sparse_vectors").entered();
        self.rt.block_on(async {
            let batch_size = load_sparse_chunk_id_batch();
            let mut result: Vec<(String, SparseVector)> = Vec::new();
            let mut total_entries: usize = 0;
            // Cursor: chunk_id ordering. Empty string sorts before any real id
            // (SQLite BINARY collation, which is the default), so using "" as
            // the initial cursor fetches the full range on the first query.
            let mut last_chunk_id = String::new();

            loop {
                // Fetch up to `batch_size` distinct chunk_ids
                // past the cursor, and all their rows, in one query. The
                // subquery's LIMIT bounds the outer rows to one batch.
                let rows: Vec<_> = sqlx::query(
                    "SELECT chunk_id, token_id, weight FROM sparse_vectors \
                     WHERE chunk_id IN ( \
                         SELECT DISTINCT chunk_id FROM sparse_vectors \
                         WHERE chunk_id > ?1 \
                         ORDER BY chunk_id \
                         LIMIT ?2 \
                     ) \
                     ORDER BY chunk_id, token_id",
                )
                .bind(&last_chunk_id)
                .bind(batch_size)
                .fetch_all(&self.pool)
                .await?;

                if rows.is_empty() {
                    break;
                }
                total_entries += rows.len();

                // Group rows by chunk_id (rows are already ORDER BY chunk_id,
                // so we can stream through linearly). Each chunk_id belongs
                // entirely to this batch — the subquery guarantees complete
                // groups, never splitting a chunk across pages.
                //
                // Borrow `chunk_id` as `&str` from the sqlx row buffer instead
                // of allocating a fresh `String` for every row. Only the
                // chunk-boundary transition needs an owned id; a 60k-chunk
                // index averaging ~100 sparse tokens per chunk would otherwise
                // allocate ~6M Strings per call (called on every watch reload +
                // daemon startup). `last_id_in_batch` keeps an owned String
                // because it crosses the loop body for the post-loop
                // dump-counting log.
                let mut current_id: Option<&str> = None;
                let mut current_vec: SparseVector = Vec::new();
                let mut last_id_in_batch = String::new();
                let mut distinct_ids_in_batch: i64 = 0;

                for row in &rows {
                    let chunk_id: &str = row.get("chunk_id");
                    let token_id: i64 = row.get("token_id");
                    let weight: f64 = row.get("weight");

                    if current_id != Some(chunk_id) {
                        if let Some(id) = current_id.take() {
                            result.push((id.to_string(), std::mem::take(&mut current_vec)));
                        }
                        last_id_in_batch.clear();
                        last_id_in_batch.push_str(chunk_id);
                        distinct_ids_in_batch += 1;
                        current_id = Some(chunk_id);
                    }
                    if token_id < 0 || token_id > u32::MAX as i64 {
                        tracing::warn!(
                            token_id,
                            chunk_id = %current_id.unwrap_or("?"),
                            "Invalid token_id, skipping"
                        );
                        continue;
                    }
                    let weight_f32 = weight as f32;
                    if !weight_f32.is_finite() {
                        tracing::warn!(
                            token_id,
                            chunk_id = %current_id.unwrap_or("?"),
                            weight = weight,
                            "Non-finite sparse-vector weight, skipping"
                        );
                        continue;
                    }
                    current_vec.push((token_id as u32, weight_f32));
                }
                if let Some(id) = current_id {
                    result.push((id.to_string(), current_vec));
                }

                // If the batch returned fewer distinct chunk_ids than the
                // page budget, the next query would find nothing — skip
                // the empty round-trip. Otherwise advance the cursor past
                // the last-seen id and loop.
                if distinct_ids_in_batch < batch_size {
                    break;
                }
                last_chunk_id = last_id_in_batch;
            }

            tracing::info!(
                chunks = result.len(),
                total_entries,
                "Sparse vectors loaded"
            );
            Ok(result)
        })
    }

    /// Get (id, text) pairs for SPLADE encoding.
    /// Text is name + signature + doc comment — the most informative NL-like fields.
    pub fn chunk_splade_texts(&self) -> Result<Vec<(String, String)>, StoreError> {
        self.chunk_splade_texts_query("SELECT id, name, signature, doc FROM chunks")
    }

    /// Like `chunk_splade_texts` but only returns chunks that don't yet have
    /// sparse vectors in the `sparse_vectors` table. For incremental indexing
    /// (CQ-4): avoids re-encoding the entire corpus on every `cqs index` run.
    pub fn chunk_splade_texts_missing(&self) -> Result<Vec<(String, String)>, StoreError> {
        self.chunk_splade_texts_query(
            "SELECT c.id, c.name, c.signature, c.doc FROM chunks c \
             WHERE c.id NOT IN (SELECT DISTINCT chunk_id FROM sparse_vectors)",
        )
    }

    fn chunk_splade_texts_query(
        &self,
        sql: &'static str,
    ) -> Result<Vec<(String, String)>, StoreError> {
        let _span = tracing::info_span!("chunk_splade_texts").entered();
        self.rt.block_on(async {
            let rows: Vec<_> = sqlx::query(sql).fetch_all(&self.pool).await?;
            let result: Vec<(String, String)> = rows
                .iter()
                .map(|row| {
                    let id: String = row.get("id");
                    let name: String = row.get("name");
                    let sig: String = row.get("signature");
                    let doc: Option<String> = row.get("doc");
                    let text = match doc {
                        Some(d) if !d.is_empty() => format!("{} {} {}", name, sig, d),
                        _ => format!("{} {}", name, sig),
                    };
                    (id, text)
                })
                .collect();
            tracing::info!(chunks = result.len(), "Loaded chunk texts for SPLADE");
            Ok(result)
        })
    }
}

impl Store<ReadWrite> {
    /// Delete sparse vectors for chunks that no longer exist.
    ///
    /// **A no-op on clean data** — the `sparse_vectors` → `chunks` FK cascade
    /// removes orphans automatically on every chunks delete. The function is
    /// kept for one-shot repair scenarios (manual SQL manipulation, restore
    /// from an older backup, data that predates the FK cascade). On a healthy
    /// DB this deletes zero rows.
    ///
    /// Wraps DELETE + generation bump in a single `begin_write` transaction so
    /// they can't run independently (which would open a lost-update race and a
    /// crash window between DELETE and the metadata UPDATE).
    pub fn prune_orphan_sparse_vectors(&self) -> Result<usize, StoreError> {
        let _span = tracing::debug_span!("prune_orphan_sparse_vectors").entered();
        self.rt.block_on(async {
            let (_guard, mut tx) = self.begin_write().await?;
            let result = sqlx::query(
                "DELETE FROM sparse_vectors WHERE chunk_id NOT IN \
                 (SELECT DISTINCT id FROM chunks)",
            )
            .execute(&mut *tx)
            .await?;
            // If any rows were actually deleted, bump the SPLADE generation
            // so stale on-disk indexes get invalidated. A prune that removes
            // zero rows leaves sparse_vectors byte-identical, so the existing
            // generation is still valid and no bump is needed.
            let affected = result.rows_affected();
            if affected > 0 {
                bump_splade_generation_tx(&mut tx).await?;
                tracing::warn!(
                    rows = affected,
                    "prune_orphan_sparse_vectors deleted rows that should have been caught by \
                     the v19 FK cascade — either FK enforcement is disabled or this database was \
                     manipulated directly. Investigate."
                );
            }
            tx.commit().await?;
            Ok(affected as usize)
        })
    }
}

impl<Mode> Store<Mode> {
    /// Read the current SPLADE generation counter from the metadata table.
    ///
    /// Returns `0` when the key is missing (fresh DB, no sparse vectors ever
    /// written, or a schema predating this field). A non-parseable value is
    /// logged as a warning and collapses to 0 — callers should treat 0 as
    /// "force rebuild" because a corrupt counter cannot be trusted.
    ///
    /// This is read on every SpladeIndex load so persisted files can be
    /// cheaply checked for staleness without walking `sparse_vectors`.
    /// A non-parseable value is warned about rather than silently collapsed to
    /// 0, so corruption stays visible; the caller still returns `0` so the
    /// loader rebuilds defensively instead of trusting a poisoned cache.
    pub fn splade_generation(&self) -> Result<u64, StoreError> {
        let _span = tracing::debug_span!("splade_generation").entered();
        self.rt.block_on(async {
            let row: Option<(String,)> =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'splade_generation'")
                    .fetch_optional(&self.pool)
                    .await?;
            Ok(match row {
                Some((s,)) => match s.parse::<u64>() {
                    Ok(n) => n,
                    Err(e) => {
                        tracing::warn!(
                            raw = %s,
                            error = %e,
                            "splade_generation metadata is not a valid u64, treating as 0 — \
                             next SPLADE load will rebuild from SQLite"
                        );
                        0
                    }
                },
                None => 0,
            })
        })
    }
}

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

    /// `load_sparse_chunk_id_batch()` returns the compiled default when
    /// unset, the env value when set, and clamps a pathological value to
    /// `i64::MAX` rather than wrapping negative.
    #[test]
    fn load_sparse_batch_honors_env() {
        std::env::remove_var("CQS_LOAD_SPARSE_BATCH");
        assert_eq!(
            load_sparse_chunk_id_batch(),
            LOAD_SPARSE_CHUNK_ID_BATCH_DEFAULT
        );
        std::env::set_var("CQS_LOAD_SPARSE_BATCH", "256");
        assert_eq!(load_sparse_chunk_id_batch(), 256);
        // Garbage / zero falls back to default (parse_env_u64 rejects them).
        std::env::set_var("CQS_LOAD_SPARSE_BATCH", "0");
        assert_eq!(
            load_sparse_chunk_id_batch(),
            LOAD_SPARSE_CHUNK_ID_BATCH_DEFAULT
        );
        // Above-i64::MAX clamps, never wraps negative.
        std::env::set_var("CQS_LOAD_SPARSE_BATCH", "18446744073709551615");
        assert_eq!(load_sparse_chunk_id_batch(), i64::MAX);
        std::env::remove_var("CQS_LOAD_SPARSE_BATCH");
    }

    /// Insert a minimal valid `chunks` row so a sparse_vectors insert can
    /// satisfy the v19 FK constraint. Pattern mirrored from
    /// `src/store/types.rs:insert_test_chunk`.
    fn insert_test_chunk(store: &Store, id: &str) {
        store.rt.block_on(async {
            let embedding = crate::embedder::Embedding::new(vec![0.0f32; crate::EMBEDDING_DIM]);
            let embedding_bytes =
                crate::store::helpers::embedding_to_bytes(&embedding, crate::EMBEDDING_DIM)
                    .unwrap();
            let now = chrono::Utc::now().to_rfc3339();
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name,
                     signature, content, content_hash, doc, line_start, line_end, embedding,
                     source_mtime, created_at, updated_at)
                     VALUES (?1, ?1, 'file', 'rust', 'function', ?1,
                     '', '', '', NULL, 1, 10, ?2, 0, ?3, ?3)",
            )
            .bind(id)
            .bind(&embedding_bytes)
            .bind(&now)
            .execute(&store.pool)
            .await
            .unwrap();
        });
    }

    fn setup_store() -> (Store, tempfile::TempDir) {
        let dir = tempfile::TempDir::new().unwrap();
        let db_path = dir.path().join("test.db");
        let store = Store::open(&db_path).unwrap();
        store.init(&crate::store::ModelInfo::default()).unwrap();
        (store, dir)
    }

    /// Insert a chunk with explicit name, signature, and optional doc.
    /// Used by chunk_splade_texts tests that need control over those fields.
    fn insert_chunk_with_fields(
        store: &Store,
        id: &str,
        name: &str,
        signature: &str,
        doc: Option<&str>,
    ) {
        store.rt.block_on(async {
            let embedding = crate::embedder::Embedding::new(vec![0.0f32; crate::EMBEDDING_DIM]);
            let embedding_bytes =
                crate::store::helpers::embedding_to_bytes(&embedding, crate::EMBEDDING_DIM)
                    .unwrap();
            let now = chrono::Utc::now().to_rfc3339();
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name,
                     signature, content, content_hash, doc, line_start, line_end, embedding,
                     source_mtime, created_at, updated_at)
                     VALUES (?1, ?1, 'file', 'rust', 'function', ?2,
                     ?3, '', '', ?4, 1, 10, ?5, 0, ?6, ?6)",
            )
            .bind(id)
            .bind(name)
            .bind(signature)
            .bind(doc)
            .bind(&embedding_bytes)
            .bind(&now)
            .execute(&store.pool)
            .await
            .unwrap();
        });
    }

    #[test]
    fn test_sparse_roundtrip() {
        let (store, _dir) = setup_store();

        // v19 requires matching chunks rows for the FK to validate.
        insert_test_chunk(&store, "chunk_a");
        insert_test_chunk(&store, "chunk_b");

        let vectors = vec![
            (
                "chunk_a".to_string(),
                vec![(1u32, 0.5f32), (2, 0.3), (3, 0.8)],
            ),
            ("chunk_b".to_string(), vec![(1, 0.7), (4, 0.6)]),
        ];

        let entries = store.upsert_sparse_vectors(&vectors).unwrap();
        assert_eq!(entries, 5); // 3 + 2

        let loaded = store.load_all_sparse_vectors().unwrap();
        assert_eq!(loaded.len(), 2);
        assert_eq!(loaded[0].0, "chunk_a");
        assert_eq!(loaded[0].1.len(), 3);
        assert_eq!(loaded[1].0, "chunk_b");
        assert_eq!(loaded[1].1.len(), 2);
    }

    #[test]
    fn test_sparse_load_filters_non_finite_weights() {
        // The load path filters non-finite weights when casting
        // `weight: f64 -> f32`. Without is_finite, an Inf/-Inf weight (corrupt
        // row, hand-edit, encoder switch) would flow into hybrid fusion as
        // f32::INFINITY and silently corrupt min-max normalization.
        // SQLite coerces NaN to NULL on bind paths so we plant Inf via
        // overflow arithmetic at insert time (NaN insertion is blocked by
        // the `weight REAL NOT NULL` column constraint via sqlx's NULL
        // mapping anyway, so the loader-side filter is the last line of
        // defense for Inf — hand-edited DBs are the realistic threat).
        let (store, _dir) = setup_store();
        insert_test_chunk(&store, "chunk_a");

        store
            .rt
            .block_on(async {
                let mut conn = store.pool.acquire().await?;
                sqlx::query(
                    "INSERT INTO sparse_vectors (chunk_id, token_id, weight) \
                     VALUES ('chunk_a', 1, 1e308 * 1e308)",
                )
                .execute(&mut *conn)
                .await?;
                sqlx::query(
                    "INSERT INTO sparse_vectors (chunk_id, token_id, weight) \
                     VALUES ('chunk_a', 2, -(1e308 * 1e308))",
                )
                .execute(&mut *conn)
                .await?;
                sqlx::query(
                    "INSERT INTO sparse_vectors (chunk_id, token_id, weight) \
                     VALUES ('chunk_a', 3, 0.5)",
                )
                .execute(&mut *conn)
                .await?;
                Ok::<(), sqlx::Error>(())
            })
            .unwrap();

        let loaded = store.load_all_sparse_vectors().unwrap();
        assert_eq!(loaded.len(), 1);
        assert_eq!(loaded[0].0, "chunk_a");
        // Only the finite (token_id=3, weight=0.5) row should survive.
        assert_eq!(loaded[0].1.len(), 1);
        assert_eq!(loaded[0].1[0].0, 3);
        assert!((loaded[0].1[0].1 - 0.5_f32).abs() < 1e-6);
    }

    #[test]
    fn test_sparse_upsert_replaces() {
        let (store, _dir) = setup_store();
        insert_test_chunk(&store, "chunk_a");

        let v1 = vec![("chunk_a".to_string(), vec![(1u32, 0.5f32)])];
        store.upsert_sparse_vectors(&v1).unwrap();

        // Upsert with different values
        let v2 = vec![("chunk_a".to_string(), vec![(2u32, 0.9f32), (3, 0.1)])];
        store.upsert_sparse_vectors(&v2).unwrap();

        let loaded = store.load_all_sparse_vectors().unwrap();
        assert_eq!(loaded.len(), 1);
        assert_eq!(loaded[0].1.len(), 2); // replaced, not appended
        assert_eq!(loaded[0].1[0].0, 2); // new token
    }

    #[test]
    fn test_sparse_empty() {
        let (store, _dir) = setup_store();
        let loaded = store.load_all_sparse_vectors().unwrap();
        assert!(loaded.is_empty());
    }

    #[test]
    fn test_splade_generation_starts_at_zero_and_is_monotonic() {
        // Audit Test Coverage (Adversarial): prune_orphan_sparse_vectors and
        // splade_generation had zero tests total. This covers the happy path
        // of the counter across upsert + prune.
        let (store, _dir) = setup_store();
        insert_test_chunk(&store, "c1");

        assert_eq!(store.splade_generation().unwrap(), 0);

        store
            .upsert_sparse_vectors(&[("c1".to_string(), vec![(1u32, 0.5f32)])])
            .unwrap();
        let after_upsert = store.splade_generation().unwrap();
        assert!(after_upsert >= 1, "upsert must bump generation");

        // Second upsert bumps again.
        store
            .upsert_sparse_vectors(&[("c1".to_string(), vec![(2u32, 0.5f32)])])
            .unwrap();
        let after_second = store.splade_generation().unwrap();
        assert!(
            after_second > after_upsert,
            "second upsert must bump generation strictly"
        );
    }

    #[test]
    fn test_prune_orphan_no_rows_does_not_bump_generation() {
        // Audit Test Coverage (Adversarial): the zero-rows-deleted branch
        // that skips the generation bump. A regression flipping the
        // condition would silently invalidate the persisted SpladeIndex
        // on every `cqs index`.
        let (store, _dir) = setup_store();
        insert_test_chunk(&store, "c1");
        store
            .upsert_sparse_vectors(&[("c1".to_string(), vec![(1u32, 0.5f32)])])
            .unwrap();

        let before = store.splade_generation().unwrap();
        let pruned = store.prune_orphan_sparse_vectors().unwrap();
        assert_eq!(pruned, 0, "v19 FK cascade should leave zero orphans");
        let after = store.splade_generation().unwrap();
        assert_eq!(
            before, after,
            "zero-orphan prune must NOT bump the generation"
        );
    }

    #[test]
    fn test_prune_orphan_on_clean_db_is_noop() {
        // The zero-rows branch must be a true no-op on a clean
        // DB that has never held any sparse vectors. A regression flipping
        // the `if affected > 0` guard to `>= 0` would bump the generation
        // on every `cqs index` even when no work was done.
        let (store, _dir) = setup_store();

        let before = store.splade_generation().unwrap();
        let pruned = store.prune_orphan_sparse_vectors().unwrap();
        assert_eq!(pruned, 0, "fresh DB must have zero orphans to prune");
        let after = store.splade_generation().unwrap();
        assert_eq!(before, after, "no-op prune on clean DB must not bump");
    }

    #[test]
    fn test_fk_cascade_removes_sparse_rows_on_chunk_delete() {
        // Audit DS-W3: v19 FK CASCADE contract. Deleting a chunk must
        // automatically remove its sparse_vectors rows.
        let (store, _dir) = setup_store();
        insert_test_chunk(&store, "c1");
        insert_test_chunk(&store, "c2");
        store
            .upsert_sparse_vectors(&[
                ("c1".to_string(), vec![(1u32, 0.5f32), (2, 0.3)]),
                ("c2".to_string(), vec![(3u32, 0.7f32)]),
            ])
            .unwrap();

        // Directly delete c1 from chunks and confirm the cascade removed
        // its sparse rows without any explicit cleanup.
        store.rt.block_on(async {
            sqlx::query("DELETE FROM chunks WHERE id = 'c1'")
                .execute(&store.pool)
                .await
                .unwrap();
        });
        let loaded = store.load_all_sparse_vectors().unwrap();
        assert_eq!(
            loaded.len(),
            1,
            "cascade should have dropped c1's sparse rows"
        );
        assert_eq!(loaded[0].0, "c2");
    }

    // ── Gap 1: chunk_splade_texts concatenation format ──────────────

    #[test]
    fn test_chunk_splade_texts_with_doc() {
        let (store, _dir) = setup_store();
        insert_chunk_with_fields(&store, "c1", "foo", "fn foo()", Some("does things"));
        let texts = store.chunk_splade_texts().unwrap();
        assert_eq!(texts.len(), 1);
        assert_eq!(texts[0].1, "foo fn foo() does things");
    }

    #[test]
    fn test_chunk_splade_texts_no_doc() {
        let (store, _dir) = setup_store();
        insert_chunk_with_fields(&store, "c1", "foo", "fn foo()", None);
        let texts = store.chunk_splade_texts().unwrap();
        assert_eq!(texts.len(), 1);
        assert_eq!(texts[0].1, "foo fn foo()");
    }

    #[test]
    fn test_chunk_splade_texts_empty_doc_treated_as_missing() {
        let (store, _dir) = setup_store();
        insert_chunk_with_fields(&store, "c1", "foo", "fn foo()", Some(""));
        let texts = store.chunk_splade_texts().unwrap();
        assert_eq!(texts.len(), 1);
        assert_eq!(
            texts[0].1, "foo fn foo()",
            "empty doc should not be appended"
        );
    }

    // ── Gap 2: chunk_splade_texts_missing ───────────────────────────

    #[test]
    fn test_chunk_splade_texts_missing_skips_encoded() {
        let (store, _dir) = setup_store();
        insert_chunk_with_fields(&store, "c1", "alpha", "fn alpha()", None);
        insert_chunk_with_fields(&store, "c2", "beta", "fn beta()", None);
        insert_chunk_with_fields(&store, "c3", "gamma", "fn gamma()", None);

        // Encode c1 and c3, leave c2 without sparse vectors.
        store
            .upsert_sparse_vectors(&[
                ("c1".to_string(), vec![(1u32, 0.5f32)]),
                ("c3".to_string(), vec![(2u32, 0.7f32)]),
            ])
            .unwrap();

        let missing = store.chunk_splade_texts_missing().unwrap();
        assert_eq!(missing.len(), 1, "only the un-encoded chunk should appear");
        assert_eq!(missing[0].0, "c2");
        assert_eq!(missing[0].1, "beta fn beta()");
    }

    // ── Gap 3: load_all_sparse_vectors grouping ─────────────────────

    #[test]
    fn test_load_all_sparse_vectors_groups_multiple_tokens() {
        let (store, _dir) = setup_store();
        insert_test_chunk(&store, "c1");
        insert_test_chunk(&store, "c2");
        insert_test_chunk(&store, "c3");

        store
            .upsert_sparse_vectors(&[
                (
                    "c1".to_string(),
                    vec![(10u32, 0.1f32), (11, 0.2), (12, 0.3)],
                ),
                (
                    "c2".to_string(),
                    vec![(20u32, 0.4f32), (21, 0.5), (22, 0.6), (23, 0.7)],
                ),
                (
                    "c3".to_string(),
                    vec![(30u32, 0.8f32), (31, 0.9), (32, 1.0)],
                ),
            ])
            .unwrap();

        let loaded = store.load_all_sparse_vectors().unwrap();
        assert_eq!(loaded.len(), 3);

        // Find each chunk and verify token counts.
        let c1 = loaded.iter().find(|(id, _)| id == "c1").unwrap();
        let c2 = loaded.iter().find(|(id, _)| id == "c2").unwrap();
        let c3 = loaded.iter().find(|(id, _)| id == "c3").unwrap();
        assert_eq!(c1.1.len(), 3, "c1 should have 3 tokens");
        assert_eq!(c2.1.len(), 4, "c2 should have 4 tokens");
        assert_eq!(c3.1.len(), 3, "c3 should have 3 tokens");
    }

    /// Mutex serializing tests that mutate `CQS_SPARSE_CHUNKS_PER_TX` —
    /// `std::env::set_var` is process-global and `cargo test` runs lib
    /// tests in parallel by default, so without serialization these
    /// tests race each other (a sibling test resets the var mid-call).
    /// Scope is deliberately just these tests; other tests don't touch
    /// this var.
    static ENV_CHUNKS_PER_TX_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());

    /// Helper: run `f` with `CQS_SPARSE_CHUNKS_PER_TX` set to `value`,
    /// restoring the prior value (or absence) afterward. Holds
    /// `ENV_CHUNKS_PER_TX_LOCK` for the duration so concurrent tests
    /// don't see a transient mid-mutation env state.
    fn with_chunks_per_tx<R>(value: &str, f: impl FnOnce() -> R) -> R {
        let _g = ENV_CHUNKS_PER_TX_LOCK
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        let prev = std::env::var("CQS_SPARSE_CHUNKS_PER_TX").ok();
        // SAFETY: the mutex above ensures no concurrent test in this binary
        // is reading or writing this env var while we mutate it.
        unsafe { std::env::set_var("CQS_SPARSE_CHUNKS_PER_TX", value) };

        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));

        unsafe {
            match prev {
                Some(v) => std::env::set_var("CQS_SPARSE_CHUNKS_PER_TX", v),
                None => std::env::remove_var("CQS_SPARSE_CHUNKS_PER_TX"),
            }
        }
        match result {
            Ok(r) => r,
            Err(payload) => std::panic::resume_unwind(payload),
        }
    }

    /// #1212: chunked sub-transactions bump `splade_generation` per
    /// committed batch so a partial-bulk crash forces `splade.index.bin`
    /// rebuild on next load. With `CQS_SPARSE_CHUNKS_PER_TX=1`, a 4-chunk
    /// upsert produces 1 (DROP step) + 4 (per-chunk bulk) + 1 (CREATE
    /// step) = 6 generation bumps. The lower-bound check pins "one bump
    /// per committed batch tx" without coupling to internal restructuring.
    #[test]
    fn test_upsert_sparse_vectors_bumps_generation_per_chunked_tx() {
        with_chunks_per_tx("1", || {
            let (store, _dir) = setup_store();
            insert_test_chunk(&store, "c1");
            insert_test_chunk(&store, "c2");
            insert_test_chunk(&store, "c3");
            insert_test_chunk(&store, "c4");

            let before = store.splade_generation().unwrap();
            store
                .upsert_sparse_vectors(&[
                    ("c1".to_string(), vec![(1u32, 0.1f32)]),
                    ("c2".to_string(), vec![(2u32, 0.2f32)]),
                    ("c3".to_string(), vec![(3u32, 0.3f32)]),
                    ("c4".to_string(), vec![(4u32, 0.4f32)]),
                ])
                .unwrap();
            let after = store.splade_generation().unwrap();
            // 1 (DROP) + 4 (bulk sub-txs) + 1 (CREATE) = 6 bumps minimum.
            assert!(
                after >= before + 6,
                "expected at least 6 bumps for 1+4+1 sub-txs (got before={before} after={after})"
            );
        });
    }

    /// `CQS_SPARSE_CHUNKS_PER_TX` invalid values fall back to the
    /// default. Guards against a typo in the env var silently producing
    /// either zero-sized batches (panic in `chunks(0)`) or non-numeric
    /// strings (parse error swallowed and ignored).
    #[test]
    fn test_upsert_sparse_vectors_env_fallback_on_invalid_value() {
        for bad_value in &["", "0", "not-a-number", "-5"] {
            with_chunks_per_tx(bad_value, || {
                let (store, _dir) = setup_store();
                insert_test_chunk(&store, "c1");
                store
                    .upsert_sparse_vectors(&[("c1".to_string(), vec![(1u32, 0.5f32)])])
                    .unwrap_or_else(|e| panic!("upsert failed for bad_value={bad_value:?}: {e}"));
                let loaded = store.load_all_sparse_vectors().unwrap();
                assert_eq!(loaded.len(), 1, "bad_value={bad_value:?} produced no rows");
            });
        }
    }

    /// The chunked path correctly handles batches that straddle
    /// CHUNKS_PER_TX boundaries — the second batch's DELETE must not
    /// affect the first batch's freshly-committed rows. Pin via:
    /// pre-seed v1 vectors for c1+c2; upsert v2 for c1+c2 with
    /// CHUNKS_PER_TX=1 (so c1 commits first, then c2). Final state must
    /// be v2 for both, not a mix.
    #[test]
    fn test_upsert_sparse_vectors_chunked_tx_preserves_atomicity_per_chunk() {
        with_chunks_per_tx("1", || {
            let (store, _dir) = setup_store();
            insert_test_chunk(&store, "c1");
            insert_test_chunk(&store, "c2");

            // Seed v1.
            store
                .upsert_sparse_vectors(&[
                    ("c1".to_string(), vec![(1u32, 0.1f32)]),
                    ("c2".to_string(), vec![(2u32, 0.2f32)]),
                ])
                .unwrap();

            // Replace with v2 (different tokens to make the swap visible).
            store
                .upsert_sparse_vectors(&[
                    ("c1".to_string(), vec![(11u32, 0.11f32), (12, 0.12)]),
                    ("c2".to_string(), vec![(21u32, 0.21f32)]),
                ])
                .unwrap();

            let loaded = store.load_all_sparse_vectors().unwrap();
            let c1 = loaded.iter().find(|(id, _)| id == "c1").unwrap();
            let c2 = loaded.iter().find(|(id, _)| id == "c2").unwrap();
            assert_eq!(c1.1.len(), 2, "c1 must hold v2's two tokens");
            assert!(
                c1.1.iter().any(|(t, _)| *t == 11),
                "c1 must contain new token 11 (v2)"
            );
            assert!(
                !c1.1.iter().any(|(t, _)| *t == 1),
                "c1 must NOT retain v1 token 1"
            );
            assert_eq!(c2.1.len(), 1, "c2 must hold v2's single token");
            assert_eq!(c2.1[0].0, 21, "c2 must contain new token 21 (v2)");
        });
    }

    /// Pin write-path behavior for non-finite weights. The read path filters
    /// NaN/Inf at load time; the write path binds the f32 directly via
    /// `push_bind`. Document the current contract so a future write-side
    /// scrubber is a deliberate change.
    ///
    /// Current behavior (pinned by this test):
    /// - `+Infinity` weight: SQLite stores `Inf`; load_all_sparse_vectors
    ///   filters it out at read time so the row is silently skipped.
    /// - `NaN` weight: SQLite's binding of NaN is implementation-defined;
    ///   may store NaN (filtered at read) or coerce. We assert the
    ///   upsert succeeds without panic and that the read-side filter
    ///   masks any non-finite values from downstream search.
    ///
    /// The "right" long-term fix is to scrub at write-time so the bytes
    /// never reach disk. This test will need flipping when that lands.
    #[test]
    fn test_upsert_sparse_vectors_inf_weight_filtered_at_read() {
        let (store, _dir) = setup_store();
        insert_test_chunk(&store, "c1");
        // Mix one finite weight with one +Inf weight.
        let result = store.upsert_sparse_vectors(&[(
            "c1".to_string(),
            vec![(1u32, 0.5f32), (2u32, f32::INFINITY)],
        )]);
        assert!(
            result.is_ok(),
            "upsert with +Inf weight currently succeeds (write-path doesn't validate); err = {:?}",
            result.err()
        );
        // Read-side filter must mask non-finite: only the finite token 1
        // survives the load.
        let loaded = store.load_all_sparse_vectors().unwrap();
        let c1 = &loaded[0];
        assert!(
            c1.1.iter().all(|(_, w)| w.is_finite()),
            "read path must filter non-finite weights: {:?}",
            c1.1
        );
        assert!(
            c1.1.iter().any(|(t, _)| *t == 1),
            "finite weight survives the round-trip"
        );
    }

    #[test]
    fn test_upsert_sparse_vectors_nan_weight_does_not_panic() {
        let (store, _dir) = setup_store();
        insert_test_chunk(&store, "c1");
        // NaN binding behavior is sqlx-dependent; some drivers coerce
        // NaN→NULL, which would violate the NOT NULL constraint and
        // surface as Err. We accept either outcome — what we pin is
        // "doesn't panic" and "if it succeeds, read-side filter masks".
        let result = store.upsert_sparse_vectors(&[("c1".to_string(), vec![(1u32, f32::NAN)])]);
        if let Ok(_count) = result {
            let loaded = store.load_all_sparse_vectors().unwrap();
            for (_id, weights) in &loaded {
                for (_, w) in weights {
                    assert!(
                        w.is_finite(),
                        "NaN must not survive the round-trip: weight = {w}"
                    );
                }
            }
        }
        // If Err, the contract is "writes reject NaN" — also acceptable
        // and pinned by the assertion that we got Ok-or-Err, not panic.
    }
}
