//! Row-level write helpers that keep the FTS5 contentless table in sync with
//! `files`/`documents_text`.
//!
//! States (mirrors the `content_state` column):
//!
//! | value | meaning |
//! |------:|---------|
//! |     0 | pending |
//! |     1 | done    |
//! |     2 | failed  |
//! |     3 | not applicable (content only) |

use rusqlite::{params, params_from_iter, Connection, OptionalExtension, Transaction};

use crate::mime::FileType;

pub const STATE_PENDING: i64 = 0;
pub const STATE_DONE: i64 = 1;
pub const STATE_FAILED: i64 = 2;
pub const STATE_NA: i64 = 3;

/// `prepare_cached` + `execute`, returning the affected row count. `what` is
/// lazy so the message is built only on the error path.
fn exec(
    conn: &Connection,
    sql: &str,
    params: impl rusqlite::Params,
    what: impl FnOnce() -> String,
) -> Result<usize, String> {
    conn.prepare_cached(sql)
        .and_then(|mut stmt| stmt.execute(params))
        .map_err(|e| format!("{}: {}", what(), e))
}

/// Set a file's content state and clear any failure record with it, as one
/// operation: `list-failed` reads `failed_files` directly, so a stale entry
/// tells the user a file is broken after it stopped being.
/// [`set_content_failed`] — the one transition that *writes* a failure
/// record — does not route through here.
fn set_state_clearing_failure(
    tx: &Transaction<'_>,
    file_id: i64,
    state: i64,
    transition: &'static str,
) -> Result<(), String> {
    exec(
        tx,
        "UPDATE files SET content_state = ?1 WHERE id = ?2",
        params![state, file_id],
        || format!("{} content_state {}", transition, file_id),
    )?;
    exec(
        tx,
        "DELETE FROM failed_files WHERE file_id = ?1",
        params![file_id],
        || format!("clear failed_files {}", file_id),
    )?;
    Ok(())
}

/// Everything needed to insert a fresh file row.
#[derive(Debug, Clone)]
pub struct NewFile<'a> {
    pub name: &'a str,
    pub path: &'a str,
    pub parent: &'a str,
    pub size: u64,
    pub mtime: u64,
    pub mime: Option<&'a str>,
    pub ftype: FileType,
    pub hash: Option<&'a [u8]>,
    /// `false` means the row is born `STATE_NA`; decided at walk time by
    /// [`crate::file_handling::content_extractable`].
    pub needs_content: bool,
}

/// Insert a new file row, returning its id. `content_state` comes from
/// `needs_content`; there is no separate basic state, because the row
/// existing *is* the basic-index state. `INSERT OR IGNORE`: a UNIQUE(path)
/// collision returns `None` rather than aborting the batch.
pub fn insert_file(tx: &Transaction<'_>, f: &NewFile<'_>) -> Result<Option<i64>, String> {
    let rows = tx
        .prepare_cached(
            "INSERT OR IGNORE INTO files (
                name, path, parent, size, mtime, mime, type, content_state, hash
             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
        )
        .and_then(|mut stmt| {
            stmt.execute(params![
                f.name,
                f.path,
                f.parent,
                f.size as i64,
                f.mtime as i64,
                f.mime,
                f.ftype.bits() as i64,
                initial_content_state(f),
                f.hash,
            ])
        })
        .map_err(|e| format!("insert file {}: {}", f.path, e))?;
    if rows == 0 {
        return Ok(None);
    }
    Ok(Some(tx.last_insert_rowid()))
}

/// The `content_state` a freshly written row starts in. Shared by
/// [`insert_file`] and [`update_file_basic`] so a file that reappears as an
/// update lands in the same state it would have as an insert.
fn initial_content_state(f: &NewFile<'_>) -> i64 {
    if f.needs_content {
        STATE_PENDING
    } else {
        STATE_NA
    }
}

/// Update a file's metadata in place (same path, changed size/mtime/hash) and
/// reset its content state from `f.needs_content`, clearing any extracted
/// content so the text-indexing pass re-processes it. Writes `size`, `mtime`,
/// `hash`, `mime`, `type` and `content_state` — and only those; `name` and
/// `parent` are not refreshed here.
pub fn update_file_basic(tx: &Transaction<'_>, f: &NewFile<'_>) -> Result<Option<i64>, String> {
    let id: Option<i64> = tx
        .prepare_cached(
            "UPDATE files
                SET size = ?1, mtime = ?2, hash = ?3, mime = ?4, type = ?5,
                    content_state = ?6
              WHERE path = ?7
          RETURNING id",
        )
        .and_then(|mut stmt| {
            stmt.query_row(
                params![
                    f.size as i64,
                    f.mtime as i64,
                    f.hash,
                    f.mime,
                    f.ftype.bits() as i64,
                    initial_content_state(f),
                    f.path,
                ],
                |r| r.get(0),
            )
            .optional()
        })
        .map_err(|e| format!("update file {}: {}", f.path, e))?;
    let Some(id) = id else {
        return Ok(None);
    };
    remove_content_for_id(tx, id)?;
    Ok(Some(id))
}

/// Mark a file's content indexing as complete and write the extracted text
/// atomically. The plaintext feeds the contentless FTS5 tokenizer; when
/// `store_text` is `true` it is also stored zstd-compressed in
/// `documents_text` for snippet rendering (`false`: matches still work, but
/// result rows can't render snippets).
/// `text_zstd` is the already-compressed body for the `documents_text`
/// sidecar, or `None` to write no sidecar at all (an empty body, or
/// `store_text_for_snippets` off).
///
/// Compression is the caller's job, and deliberately so: it is the expensive
/// half of a content write, and this runs inside the writer's transaction
/// with the shared connection held. Callers on the indexing path compress a
/// whole batch through one [`DocEncoder`] *before* taking the lock, so the
/// transaction only binds finished blobs.
pub fn set_content_done(
    tx: &Transaction<'_>,
    file_id: i64,
    text: &str,
    text_zstd: Option<&[u8]>,
) -> Result<(), String> {
    remove_content_for_id(tx, file_id)?;

    // Contentless FTS5 still accepts values on INSERT — the tokenizer needs
    // them — it simply doesn't persist the raw column values.
    exec(
        tx,
        "INSERT INTO searchabletext(rowid, text) VALUES (?1, ?2)",
        params![file_id, text],
        || format!("insert FTS row {}", file_id),
    )?;

    // No sidecar row for empty body text (an audio file whose tags are all
    // empty, say) — the caller passes `None` for that.
    if let Some(compressed) = text_zstd {
        exec(
            tx,
            "INSERT INTO documents_text(file_id, text_zstd) VALUES (?1, ?2)",
            params![file_id, compressed],
            || format!("insert documents_text {}", file_id),
        )?;
    }

    set_state_clearing_failure(tx, file_id, STATE_DONE, "update DONE")
}

/// Reusable decode buffer and decompression context for the readers of
/// `documents_text` — the read side's mirror of [`DocEncoder`].
///
/// Shared by the cascade's full-text passes and by [`crate::live`], which
/// re-reads one row's body when a file under a visible result changes.
///
/// `zstd::decode_all` builds and tears down a `ZSTD_DCtx` *and* allocates a
/// fresh output `Vec` on every call, and it is called once per candidate row.
/// One context and one buffer, reused across a whole scan, make that a
/// per-scan cost instead of a per-row one.
pub struct DocDecoder {
    dctx: zstd::bulk::Decompressor<'static>,
    buf: Vec<u8>,
}

/// Where [`DocDecoder::decode`]'s buffer starts before it has seen a document.
/// Most extracted text is well under this, so the doubling below rarely runs.
const INITIAL_DOC_CAPACITY: usize = 64 * 1024;

/// Where the doubling stops. Stored text is capped at
/// `processing.maximum_text_size` (256 KiB by default), so this is far above
/// any legitimate document even if that setting is raised — past it, a failure
/// is a corrupt frame rather than a buffer that is too small.
const MAX_DOC_CAPACITY: usize = 64 * 1024 * 1024;

impl DocDecoder {
    pub fn new() -> Result<Self, String> {
        Ok(DocDecoder {
            dctx: zstd::bulk::Decompressor::new().map_err(|e| e.to_string())?,
            buf: Vec::new(),
        })
    }

    /// Decompress `blob` and borrow the result as text.
    ///
    /// Returns `None` for a corrupt frame or non-UTF-8 content. Nothing is
    /// copied: the indexer stores UTF-8, so the bytes are borrowed in place
    /// rather than run through `String::from_utf8_lossy(..).into_owned()`,
    /// which duplicated the whole document even when it was already valid.
    pub fn decode(&mut self, blob: &[u8]) -> Option<&str> {
        self.buf.clear();
        // `decompress_to_buffer` writes into spare capacity and fails rather
        // than growing, so the room has to be there first.
        //
        // [`DocEncoder`] compresses through `ZSTD_compress2`, which knows the
        // whole input up front and records its length in the frame header, so
        // this reservation is normally exact and the loop below runs once.
        //
        // The loop is still the fallback, and it is not optional: a frame
        // written by a *stream*-based encoder carries no content size, and
        // falling back to `zstd::decode_all` for those looked harmless and was
        // not — it builds a streaming decoder per call, which measured as one
        // ~2.4 MiB allocation per document and 27 of the 30 GiB a fuzzy search
        // moved through the allocator. Growing and reusing this buffer instead
        // settles at the largest document in the scan within the first few
        // rows, after which decoding a row allocates nothing at all.
        if let Ok(Some(size)) = zstd::zstd_safe::get_frame_content_size(blob) {
            self.buf.reserve(usize::try_from(size).ok()?);
        }
        loop {
            if self.buf.capacity() == 0 {
                self.buf.reserve(INITIAL_DOC_CAPACITY);
            }
            match self.dctx.decompress_to_buffer(blob, &mut self.buf) {
                Ok(_) => break,
                // Too small, or corrupt — the bulk API cannot tell us which.
                // Growing is only worth trying while the buffer is still
                // smaller than any document could legitimately be.
                Err(_) if self.buf.capacity() < MAX_DOC_CAPACITY => {
                    let bigger = self.buf.capacity().saturating_mul(2);
                    self.buf.clear();
                    self.buf.reserve(bigger);
                }
                Err(_) => return None,
            }
        }
        std::str::from_utf8(&self.buf).ok()
    }
}

/// Level 3 hits ~3-5× on English prose at high throughput (hundreds of
/// MB/s); level 9+ would shave a few percent more at 10× the CPU cost, and
/// readers decompress far faster than writers compress.
const ZSTD_LEVEL: i32 = 3;

/// Reusable compression context for the `documents_text` sidecar — the write
/// side's mirror of the cascade's `DocDecoder`.
///
/// `zstd::encode_all` builds and tears down a `ZSTD_CCtx` — window, hash and
/// chain tables — on every call, and the writer calls it once per extracted
/// document. At 1 KiB, the size most documents actually are, that setup costs
/// more than the compression: 16.4 µs against 3.4 µs for the same bytes
/// through a context that already exists (`benches/index.rs`, group
/// `zstd_encode`). One encoder per batch makes it a per-batch cost.
pub struct DocEncoder(zstd::bulk::Compressor<'static>);

impl DocEncoder {
    pub fn new() -> Result<DocEncoder, String> {
        zstd::bulk::Compressor::new(ZSTD_LEVEL)
            .map(DocEncoder)
            .map_err(|e| format!("zstd encoder: {}", e))
    }

    /// Compress `text` for [`set_content_done`]'s `text_zstd` argument.
    pub fn encode(&mut self, text: &str) -> Result<Vec<u8>, String> {
        self.0
            .compress(text.as_bytes())
            .map_err(|e| format!("zstd encode: {}", e))
    }
}

/// Compress one body, for the writers that handle a single row.
///
/// The batch writers reuse one [`DocEncoder`] across a chunk and run it
/// outside the connection lock. The single-row paths — the watcher, and a
/// walk-time inline body — write one row per transaction, so there is no
/// batch to amortize a context over and this builds one for the document.
pub fn encode_one(text: &str, store_text: bool) -> Result<Option<Vec<u8>>, String> {
    if !store_text || text.is_empty() {
        return Ok(None);
    }
    DocEncoder::new()?.encode(text).map(Some)
}

/// The uncompressed size of a stored `documents_text` blob, read out of the
/// zstd frame header instead of from a column beside it.
///
/// [`DocEncoder`] compresses through `ZSTD_compress2`, which is handed the
/// whole document at once and writes its length into the frame header. A
/// `text_len` column would have stored that same number a second time for
/// every row, to serve one figure in the size report.
///
/// `None` for a frame that carries no content size — nothing this writer
/// produces — or a corrupt one. The caller only needs the frame *header*, so
/// `blob` may be a prefix of the stored value.
pub fn raw_text_len(blob: &[u8]) -> Option<u64> {
    zstd::zstd_safe::get_frame_content_size(blob).ok().flatten()
}

/// Mark a file's content extraction as failed. Keeps the basic row in place.
///
/// The reason is written once, to `failed_files` — which also carries the
/// timestamp, and which `list-failed` and `status` both read.
pub fn set_content_failed(tx: &Transaction<'_>, file_id: i64, reason: &str) -> Result<(), String> {
    let now = crate::log::now_unix() as i64;
    exec(
        tx,
        "UPDATE files SET content_state = ?1 WHERE id = ?2",
        params![STATE_FAILED, file_id],
        || format!("update content_state FAILED {}", file_id),
    )?;
    exec(
        tx,
        "INSERT OR REPLACE INTO failed_files(file_id, reason, ts) VALUES (?1, ?2, ?3)",
        params![file_id, reason, now],
        || format!("insert failed_files {}", file_id),
    )?;
    Ok(())
}

/// Mark content extraction as not applicable (e.g. binary format we don't
/// support). The file row still contributes to filename search.
pub fn set_content_na(tx: &Transaction<'_>, file_id: i64) -> Result<(), String> {
    set_state_clearing_failure(tx, file_id, STATE_NA, "update NA")
}

/// Delete a file row by path, keeping FTS in sync. Returns whether a row was
/// removed.
pub fn delete_file_by_path(tx: &Transaction<'_>, path: &str) -> Result<bool, String> {
    let id: Option<i64> = tx
        .prepare_cached("DELETE FROM files WHERE path = ?1 RETURNING id")
        .and_then(|mut stmt| stmt.query_row(params![path], |r| r.get(0)).optional())
        .map_err(|e| format!("delete file {}: {}", path, e))?;
    let Some(id) = id else { return Ok(false) };
    remove_content_for_id(tx, id)?;
    Ok(true)
}

/// Delete every row whose path falls in the half-open range `[lo, hi)`,
/// keeping the dependent tables in step. Returns how many `files` rows went.
///
/// Five statements regardless of how many files the range holds, and the
/// range is an index seek on `UNIQUE(files.path)`. Build the bounds with
/// [`crate::file_handling::ExtractCursor::for_root`], which is what makes
/// them separator-correct.
pub fn delete_subtree(tx: &Transaction<'_>, lo: &str, hi: &str) -> Result<usize, String> {
    for (table, key) in DEPENDENT_TABLES {
        let sql = format!(
            "DELETE FROM {} WHERE {} IN \
             (SELECT id FROM files WHERE path >= ?1 AND path < ?2)",
            table, key
        );
        exec(tx, &sql, params![lo, hi], || {
            format!("delete {} under {}", table, lo)
        })?;
    }
    exec(
        tx,
        "DELETE FROM files WHERE path >= ?1 AND path < ?2",
        params![lo, hi],
        || format!("delete files under {}", lo),
    )
}

/// Delete every row whose path falls in *none* of `ranges`. Returns how many
/// `files` rows went.
///
/// A scan of `files` rather than a seek, reserved for the one transition that
/// needs it: with `follow_symlinks` off, rows left by a followed symlink fall
/// outside every root's range and no walk will ever visit them again. An
/// empty `ranges` (no roots configured — e.g. a half-written config) deletes
/// nothing.
pub fn delete_outside_ranges(
    tx: &Transaction<'_>,
    ranges: &[(String, String)],
) -> Result<usize, String> {
    if ranges.is_empty() {
        return Ok(0);
    }
    let mut predicate = String::new();
    for i in 0..ranges.len() {
        if i > 0 {
            predicate.push_str(" AND ");
        }
        predicate.push_str(&format!(
            "NOT (path >= ?{} AND path < ?{})",
            i * 2 + 1,
            i * 2 + 2
        ));
    }
    let bounds: Vec<&String> = ranges.iter().flat_map(|(lo, hi)| [lo, hi]).collect();
    for (table, key) in DEPENDENT_TABLES {
        let sql = format!(
            "DELETE FROM {} WHERE {} IN (SELECT id FROM files WHERE {})",
            table, key, predicate
        );
        exec(tx, &sql, params_from_iter(bounds.iter()), || {
            format!("delete {} outside the roots", table)
        })?;
    }
    let sql = format!("DELETE FROM files WHERE {}", predicate);
    exec(tx, &sql, params_from_iter(bounds.iter()), || {
        "delete files outside the roots".to_string()
    })
}

/// The tables a file id owns, in the order they must be cleared: everything
/// keyed to `files.id` first, then `files` itself. Not left to `ON DELETE
/// CASCADE`: `searchabletext` is an FTS5 virtual table with no foreign key at
/// all, and cascade only fires on connections with `PRAGMA foreign_keys` on.
const DEPENDENT_TABLES: [(&str, &str); 3] = [
    ("searchabletext", "rowid"),
    ("documents_text", "file_id"),
    ("failed_files", "file_id"),
];

/// How many ids [`delete_ids`] binds into one statement — fixed so
/// `prepare_cached` sees a bounded set of distinct SQL texts.
const DELETE_IDS_CHUNK: usize = 512;

/// `?,?,…` for an `IN (…)` clause binding `n` values.
fn placeholders(n: usize) -> String {
    vec!["?"; n].join(",")
}

/// Delete the given file ids and everything keyed to them. Returns how many
/// `files` rows went. For rows chosen by a predicate no SQL range can express
/// (a glob ignore pattern, say); five statements per [`DELETE_IDS_CHUNK`] ids
/// rather than five per file.
pub fn delete_ids(tx: &Transaction<'_>, ids: &[i64]) -> Result<usize, String> {
    let mut removed = 0;
    for chunk in ids.chunks(DELETE_IDS_CHUNK) {
        let placeholders = placeholders(chunk.len());
        for (table, key) in DEPENDENT_TABLES {
            let sql = format!("DELETE FROM {} WHERE {} IN ({})", table, key, placeholders);
            exec(tx, &sql, params_from_iter(chunk.iter()), || {
                format!("delete {} for {} ids", table, chunk.len())
            })?;
        }
        let sql = format!("DELETE FROM files WHERE id IN ({})", placeholders);
        removed += exec(tx, &sql, params_from_iter(chunk.iter()), || {
            format!("delete {} file rows", chunk.len())
        })?;
    }
    Ok(removed)
}

/// Every indexed file directly inside `parent`, as `name -> mtime`. Served by
/// `idx_files_parent`: one index range lookup.
pub fn dir_rows(
    conn: &Connection,
    parent: &str,
) -> Result<std::collections::HashMap<String, u64>, String> {
    let mut stmt = conn
        .prepare_cached("SELECT name, mtime FROM files WHERE parent = ?1")
        .map_err(|e| format!("prepare dir rows for {}: {}", parent, e))?;
    let rows = stmt
        .query_map(params![parent], |r| {
            Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)?.max(0) as u64))
        })
        .map_err(|e| format!("query dir rows for {}: {}", parent, e))?;
    let mut out = std::collections::HashMap::new();
    for row in rows {
        let (name, mtime) = row.map_err(|e| format!("read dir row under {}: {}", parent, e))?;
        out.insert(name, mtime);
    }
    Ok(out)
}

/// A row the content pass has yet to extract: `(id, name, path, mime)`.
pub type PendingContentRow = (i64, String, String, Option<String>);

/// One page of rows still awaiting content extraction under `cursor`'s range,
/// ordered by id.
///
/// Keyset, not `OFFSET`: each page is an index seek, and because the cursor
/// only moves forward a row is served exactly once even though the writer is
/// concurrently flipping `content_state` behind the reader.
pub fn pending_content_page(
    conn: &Connection,
    cursor: &crate::file_handling::ExtractCursor,
    max_size: i64,
    limit: i64,
) -> Result<Vec<PendingContentRow>, String> {
    let mut stmt = conn
        .prepare_cached(
            "SELECT id, name, path, mime FROM files
              WHERE content_state = 0 AND size <= ?1 AND id > ?2
                AND path >= ?3 AND path < ?4
              ORDER BY id
              LIMIT ?5",
        )
        .map_err(|e| format!("prepare pending content query: {}", e))?;
    let rows = stmt
        .query_map(
            params![max_size, cursor.last_id, cursor.lo, cursor.hi, limit],
            |row| {
                Ok((
                    row.get::<_, i64>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, Option<String>>(3)?,
                ))
            },
        )
        .map_err(|e| format!("query pending content: {}", e))?;
    rows.collect::<Result<Vec<_>, _>>()
        .map_err(|e| format!("read pending content row: {}", e))
}

/// A stored row as the scope reconciler sees it: enough to decide both
/// whether the path is still in scope and whether its content still is.
#[derive(Debug, Clone)]
pub struct ScopeRow {
    pub id: i64,
    pub path: String,
    pub size: u64,
    pub mime: Option<String>,
    pub content_state: i64,
}

/// How many files the index holds.
pub fn row_count(conn: &Connection) -> Result<usize, String> {
    conn.query_row("SELECT COUNT(*) FROM files", [], |r| r.get::<_, i64>(0))
        .map(|n| n.max(0) as usize)
        .map_err(|e| format!("count indexed files: {}", e))
}

/// What one root holds: rows under it, and how many of those are searchable
/// by content.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RootCounts {
    pub files: i64,
    /// Rows carrying a `searchabletext` entry.
    pub fts: i64,
}

/// Count the rows in the half-open path range `[lo, hi)` and, in the same
/// pass, how many of them have a full-text row.
///
/// `content_state = STATE_DONE` *is* "has a `searchabletext` row":
/// [`set_content_done`] holds the only insert into that table and is what
/// writes the state, and [`remove_content_for_id`] clears the two together.
/// Asking `files` is what makes both figures one statement — the FTS table is
/// contentless and keyed by `rowid`, so it has no path to range-scan on.
///
/// One statement, but not a cheap one: `content_state` is not carried by the
/// `UNIQUE(files.path)` index the range seeks on, so every row in the range is
/// fetched. Call it where a run has just read those rows anyway, not on a
/// cadence.
pub fn count_root(conn: &Connection, lo: &str, hi: &str) -> Result<RootCounts, String> {
    conn.prepare_cached(
        "SELECT COUNT(*), COALESCE(SUM(content_state = ?3), 0) FROM files
          WHERE path >= ?1 AND path < ?2",
    )
    .and_then(|mut stmt| {
        stmt.query_row(params![lo, hi, STATE_DONE], |r| {
            Ok(RootCounts {
                files: r.get(0)?,
                fts: r.get(1)?,
            })
        })
    })
    .map_err(|e| format!("count root {}: {}", lo, e))
}

/// One page of rows whose path is `> after` and `< hi`, in path order.
///
/// Keyset on `path`: every page is an index walk with no sort step, and a row
/// is served at most once even though the caller is deleting behind the
/// reader. Seed `after` with the range's `lo` bound, which is
/// `root + separator` and so can never equal a stored path.
pub fn rows_in_range_page(
    conn: &Connection,
    after: &str,
    hi: &str,
    limit: i64,
) -> Result<Vec<ScopeRow>, String> {
    let mut stmt = conn
        .prepare_cached(
            "SELECT id, path, size, mime, content_state FROM files
              WHERE path > ?1 AND path < ?2
              ORDER BY path
              LIMIT ?3",
        )
        .map_err(|e| format!("prepare range page: {}", e))?;
    let rows = stmt
        .query_map(params![after, hi, limit], |row| {
            Ok(ScopeRow {
                id: row.get(0)?,
                path: row.get(1)?,
                size: row.get::<_, i64>(2)?.max(0) as u64,
                mime: row.get(3)?,
                content_state: row.get(4)?,
            })
        })
        .map_err(|e| format!("query range page after {}: {}", after, e))?;
    rows.collect::<Result<Vec<_>, _>>()
        .map_err(|e| format!("read range page row: {}", e))
}

/// Drop the stored text of the given file ids, leaving their FTS row and
/// `files` row intact: full-text search keeps working, only the
/// snippet/occurrence source goes away.
pub fn drop_stored_text(tx: &Transaction<'_>, ids: &[i64]) -> Result<usize, String> {
    let mut removed = 0;
    for chunk in ids.chunks(DELETE_IDS_CHUNK) {
        let sql = format!(
            "DELETE FROM documents_text WHERE file_id IN ({})",
            placeholders(chunk.len())
        );
        removed += exec(tx, &sql, params_from_iter(chunk.iter()), || {
            format!("drop stored text for {} ids", chunk.len())
        })?;
    }
    Ok(removed)
}

/// Put a file's content back in the pending queue without touching its row's
/// metadata. For a config change that widens what gets extracted: the file
/// itself has not changed, but its content must be produced again.
pub fn reset_content_pending(tx: &Transaction<'_>, file_id: i64) -> Result<(), String> {
    remove_content_for_id(tx, file_id)?;
    set_state_clearing_failure(tx, file_id, STATE_PENDING, "reset pending")
}

/// The stored mtime for one exact path, or `None` if it isn't indexed. For
/// files whose parent isn't the directory being read (a resolved symlink
/// target), where [`dir_rows`] would not have them.
pub fn mtime_for_path(conn: &Connection, path: &str) -> Result<Option<u64>, String> {
    let mut stmt = conn
        .prepare_cached("SELECT mtime FROM files WHERE path = ?1")
        .map_err(|e| format!("prepare mtime lookup for {}: {}", path, e))?;
    stmt.query_row(params![path], |r| r.get::<_, i64>(0))
        .optional()
        .map(|o| o.map(|m| m.max(0) as u64))
        .map_err(|e| format!("mtime lookup for {}: {}", path, e))
}

/// Distinct `parent` values within the half-open path range `[lo, hi)`,
/// streamed to `f` so nothing proportional to the tree is materialized.
/// `idx_files_parent` makes this an index-only scan.
pub fn for_each_parent_in_range<F: FnMut(String)>(
    conn: &Connection,
    lo: &str,
    hi: &str,
    mut f: F,
) -> Result<(), String> {
    let mut stmt = conn
        .prepare("SELECT DISTINCT parent FROM files WHERE parent >= ?1 AND parent < ?2")
        .map_err(|e| format!("prepare parent scan: {}", e))?;
    let rows = stmt
        .query_map(params![lo, hi], |r| r.get::<_, String>(0))
        .map_err(|e| format!("parent scan: {}", e))?;
    for row in rows {
        f(row.map_err(|e| format!("read parent row: {}", e))?);
    }
    Ok(())
}

/// Paths of every file directly inside `parent`.
pub fn paths_in_dir(conn: &Connection, parent: &str) -> Result<Vec<String>, String> {
    let mut stmt = conn
        .prepare_cached("SELECT path FROM files WHERE parent = ?1")
        .map_err(|e| format!("prepare paths in {}: {}", parent, e))?;
    let rows = stmt
        .query_map(params![parent], |r| r.get::<_, String>(0))
        .map_err(|e| format!("query paths in {}: {}", parent, e))?;
    rows.collect::<Result<Vec<_>, _>>()
        .map_err(|e| format!("read path under {}: {}", parent, e))
}

/// Remove the FTS row and the compressed text blob for a given file id. Does
/// not touch the `files` row itself. Idempotent — a missing row is fine.
pub fn remove_content_for_id(tx: &Transaction<'_>, file_id: i64) -> Result<(), String> {
    // `contentless_delete=1` on the FTS5 table makes this work without
    // re-supplying the old column values (it tombstones the rowid).
    //
    // Spelled out rather than built from a (table, key) table: this runs for
    // every extracted document and every changed file, and `format!`ing two
    // constant strings per call also handed `prepare_cached` two freshly
    // allocated keys to hash.
    for (what, sql) in [
        (
            "searchabletext",
            "DELETE FROM searchabletext WHERE rowid = ?1",
        ),
        (
            "documents_text",
            "DELETE FROM documents_text WHERE file_id = ?1",
        ),
    ] {
        exec(tx, sql, params![file_id], || {
            format!("delete {} for {}", what, file_id)
        })?;
    }
    Ok(())
}

/// Free pages, as a percentage of the file, that make a [`maintain`] VACUUM
/// worth its cost: rewriting a multi-gigabyte index to reclaim a few
/// megabytes is minutes of I/O for no gain.
const VACUUM_MIN_SLACK_PERCENT: i64 = 20;

/// Flush the whole WAL into the main database and truncate the log to zero
/// bytes. `Err` means the log was *not* emptied.
///
/// `execute`/`execute_batch` discard the pragma's result row, and that row is
/// the only place SQLite reports that a checkpoint gave up — an incomplete
/// checkpoint is not an error, it is a number in a row nobody read. That is
/// how a log can grow past the index it journals without a word in the logs.
///
/// The signal is the *log* column (frames left in the WAL), not `busy`: a
/// TRUNCATE that cannot take the writer lock silently downgrades itself to
/// PASSIVE and still reports `busy = 0` with the log untouched. A real restart
/// sets `mxFrame` to 0. A database not in WAL mode reports -1, hence `<= 0`.
pub fn checkpoint_truncate(conn: &Connection) -> Result<(), String> {
    let (busy, log): (i64, i64) = conn
        .query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |r| {
            Ok((r.get(0)?, r.get(1)?))
        })
        .map_err(|e| format!("wal checkpoint: {}", e))?;
    if log <= 0 {
        Ok(())
    } else {
        Err(format!(
            "wal checkpoint incomplete: busy={}, {} frames left in the log",
            busy, log
        ))
    }
}

/// Flush the WAL into the main DB file and close. Call on clean shutdown so
/// the next open starts with an empty log. WAL mode itself is persistent in
/// the file and left on.
pub fn checkpoint_and_close(conn: Connection) {
    if let Err(e) = checkpoint_truncate(&conn) {
        crate::log_warn!("{}", e);
    }
    drop(conn);
}

/// Land the log, reclaim the file's slack, and refresh the query planner's
/// statistics. Returns whether it vacuumed.
///
/// The sequence is checkpoint → VACUUM → `PRAGMA optimize` → checkpoint. The
/// trailing checkpoint is not a repeat of the first: VACUUM's copy-back pushes
/// every page of the rebuilt file through the log, and `optimize` writes
/// `sqlite_stat1`, so leaving without one would trade the slack just reclaimed
/// for a log the size of the index.
///
/// Run on a connection from [`crate::db::open::open_maintenance`], never the
/// indexer's (see [`super::schema::PRAGMAS_MAINTENANCE`]).
///
/// `db_dir` is where the temporary database goes, and it must be the index's
/// own directory: `temp_store = FILE` alone resolves through `SQLITE_TMPDIR`,
/// `TMPDIR`, `/var/tmp`, then `/tmp` — and `/tmp` is a RAM-backed tmpfs on
/// many Linux systems.
///
/// Peak transient space on that volume is roughly three times the index: the
/// original, the replacement being built beside it, and the log that VACUUM's
/// copy-back runs through. Running out is a failed VACUUM, not a damaged
/// index — the transaction rolls back.
pub fn maintain(conn: &Connection, db_dir: &str) -> Result<bool, String> {
    // Best-effort: compaction does not need the log empty to start.
    if let Err(e) = checkpoint_truncate(conn) {
        crate::log_warn!("{}", e);
    }

    let page_count: i64 = conn
        .query_row("PRAGMA page_count", [], |r| r.get(0))
        .map_err(|e| format!("read page_count: {}", e))?;
    let freelist: i64 = conn
        .query_row("PRAGMA freelist_count", [], |r| r.get(0))
        .map_err(|e| format!("read freelist_count: {}", e))?;

    let vacuumed = freelist * 100 >= page_count * VACUUM_MIN_SLACK_PERCENT;
    if vacuumed {
        // `temp_store_directory` is a deprecated pragma that writes a global,
        // so it is set for the VACUUM and cleared straight after rather than
        // left standing for every other connection in the process.
        let escaped = db_dir.replace('\'', "''");
        conn.execute_batch(&format!("PRAGMA temp_store_directory = '{}';", escaped))
            .map_err(|e| format!("set temp dir for vacuum: {}", e))?;
        let outcome = conn
            .execute_batch("VACUUM;")
            .map_err(|e| format!("vacuum: {}", e));
        let _ = conn.execute_batch("PRAGMA temp_store_directory = '';");
        outcome?;
    }

    // Re-analyses only the tables whose shape has drifted far enough to
    // matter, so it is close to free on a run that changed little.
    conn.execute_batch("PRAGMA optimize;")
        .map_err(|e| format!("optimize: {}", e))?;

    checkpoint_truncate(conn)?;
    Ok(vacuumed)
}

/// Read the `last_full_index` marker (unix seconds of the last *successful*
/// full indexing run) from `schema_info`. Absent key — fresh DB, or a DB
/// from before this marker existed — means "never".
pub fn get_last_full_index(conn: &Connection) -> Option<u64> {
    conn.query_row(
        "SELECT value FROM schema_info WHERE key = 'last_full_index'",
        [],
        |r| r.get::<_, String>(0),
    )
    .optional()
    .ok()
    .flatten()
    .and_then(|v| v.parse().ok())
}

/// Stamp `last_full_index` with `ts` (unix seconds). Called at the end of
/// every successful full indexing run; the coordinator reads it to schedule
/// periodic reindexing.
pub fn set_last_full_index(conn: &Connection, ts: u64) -> Result<(), String> {
    conn.execute(
        "INSERT OR REPLACE INTO schema_info(key, value) VALUES ('last_full_index', ?1)",
        params![ts.to_string()],
    )
    .map_err(|e| format!("write last_full_index: {}", e))?;
    Ok(())
}

/// The `schema_info` key prefixes holding per-root figures. Every one of them
/// is swept by [`prune_root_stats`], so a new prefix belongs in this list or a
/// de-configured root leaves it behind forever.
const ROOT_STAT_PREFIXES: [&str; 2] = ["walk_count:", "counts:"];

/// `schema_info` key holding one root's figure of the given kind.
fn root_key(prefix: &str, root: &str) -> String {
    format!("{}{}", prefix, root)
}

/// `schema_info` key holding one root's last known file count.
fn walk_count_key(root: &str) -> String {
    root_key(ROOT_STAT_PREFIXES[0], root)
}

/// `schema_info` key holding one root's last completed run's [`RootCounts`].
fn counts_key(root: &str) -> String {
    root_key(ROOT_STAT_PREFIXES[1], root)
}

/// How many files the last clean walk of `root` reported — the progress bar's
/// denominator. Absent means the root has never been walked to completion.
///
/// Last run's *file* count rather than a tree-entry count: entry counts
/// include directories and ignore-pruned subtrees and so read high (over 1.6x
/// on a home directory). See
/// [`crate::indexing::RootProgress::walk_denominator`].
pub fn get_root_walk_count(conn: &Connection, root: &str) -> Option<usize> {
    conn.query_row(
        "SELECT value FROM schema_info WHERE key = ?1",
        params![walk_count_key(root)],
        |r| r.get::<_, String>(0),
    )
    .optional()
    .ok()
    .flatten()
    .and_then(|v| v.parse().ok())
}

/// Record `n` as `root`'s file count, for the next run's progress bar.
/// Written only after a walk that finished cleanly: a partial walk's count
/// would leave every later run dividing by a number that is too small.
pub fn set_root_walk_count(conn: &Connection, root: &str, n: usize) -> Result<(), String> {
    conn.execute(
        "INSERT OR REPLACE INTO schema_info(key, value) VALUES (?1, ?2)",
        params![walk_count_key(root), n.to_string()],
    )
    .map_err(|e| format!("write walk count for {}: {}", root, e))?;
    Ok(())
}

/// What the last completed run counted under `root`, if one has finished
/// since the root was configured. Absent — never indexed, cleared, or a
/// value this build cannot parse — reads as `None`, like the walk count.
pub fn get_root_counts(conn: &Connection, root: &str) -> Option<RootCounts> {
    let stored: String = conn
        .query_row(
            "SELECT value FROM schema_info WHERE key = ?1",
            params![counts_key(root)],
            |r| r.get(0),
        )
        .optional()
        .ok()
        .flatten()?;
    let (files, fts) = stored.split_once(',')?;
    Some(RootCounts {
        files: files.parse().ok()?,
        fts: fts.parse().ok()?,
    })
}

/// Record what `root` holds, for the folder list to show once the run that
/// counted it is over.
///
/// Written only at the end of a run that completed: a stopped one has counted
/// part of a tree it was still changing, and the previous figure is closer to
/// the truth than that.
pub fn set_root_counts(conn: &Connection, root: &str, counts: RootCounts) -> Result<(), String> {
    conn.execute(
        "INSERT OR REPLACE INTO schema_info(key, value) VALUES (?1, ?2)",
        params![counts_key(root), format!("{},{}", counts.files, counts.fts)],
    )
    .map_err(|e| format!("write counts for {}: {}", root, e))?;
    Ok(())
}

/// Forget the stored figures of roots that are no longer configured, so a
/// root removed and later re-added does not start from stale ones.
pub fn prune_root_stats(conn: &Connection, keep: &[String]) -> Result<(), String> {
    let keep: std::collections::HashSet<String> = ROOT_STAT_PREFIXES
        .iter()
        .flat_map(|prefix| keep.iter().map(move |r| root_key(prefix, r)))
        .collect();
    // Filtered here rather than with a `LIKE` per prefix: `schema_info` holds
    // a handful of keys plus these, and a SQL pattern list would be a second
    // spelling of `ROOT_STAT_PREFIXES` to keep in step with the first.
    let mut stmt = conn
        .prepare("SELECT key FROM schema_info")
        .map_err(|e| format!("read root stats: {}", e))?;
    let stored: Vec<String> = stmt
        .query_map([], |r| r.get::<_, String>(0))
        .map_err(|e| format!("read root stats: {}", e))?
        .filter_map(|r| r.ok())
        .filter(|k| ROOT_STAT_PREFIXES.iter().any(|p| k.starts_with(p)))
        .collect();
    drop(stmt);
    for key in stored.iter().filter(|k| !keep.contains(*k)) {
        conn.execute("DELETE FROM schema_info WHERE key = ?1", params![key])
            .map_err(|e| format!("drop root stat {}: {}", key, e))?;
    }
    Ok(())
}

#[cfg(test)]
#[path = "repo_tests.rs"]
mod tests;