pxh 0.9.24

use std::{
    collections::HashMap,
    env,
    fmt::Write as FmtWrite,
    fs::File,
    io,
    io::{BufReader, BufWriter, Read, Write as IoWrite},
    os::unix::{
        ffi::{OsStrExt, OsStringExt},
        fs::MetadataExt,
    },
    path::{Path, PathBuf},
    str,
    sync::Arc,
    time::{Duration, Instant},
};

use bstr::{BString, ByteSlice, io::BufReadExt};
use chrono::prelude::{Local, TimeZone};
use itertools::Itertools;
use regex::bytes::Regex;
use rusqlite::{
    Connection, Error, ErrorCode, Result, Row, Transaction, TransactionBehavior,
    functions::FunctionFlags,
};
use serde::{Deserialize, Serialize};

type BoxError = Box<dyn std::error::Error + Send + Sync + 'static>;

pub mod recall;
pub mod secrets_patterns;

pub fn get_setting(
    conn: &Connection,
    key: &str,
) -> Result<Option<BString>, Box<dyn std::error::Error>> {
    let mut stmt = conn.prepare("SELECT value FROM settings WHERE key = ?")?;
    let mut rows = stmt.query([key])?;

    if let Some(row) = rows.next()? {
        let value: Vec<u8> = row.get(0)?;
        Ok(Some(BString::from(value)))
    } else {
        Ok(None)
    }
}

pub fn set_setting(
    conn: &Connection,
    key: &str,
    value: &BString,
) -> Result<(), Box<dyn std::error::Error>> {
    conn.execute(
        "INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)",
        (key, value.as_bytes()),
    )?;
    Ok(())
}

const TIME_FORMAT: &str = "%Y-%m-%d %H:%M:%S";

pub fn get_hostname() -> BString {
    let hostname =
        env::var_os("PXH_HOSTNAME").unwrap_or_else(|| hostname::get().unwrap_or_default());

    // Extract short hostname (before first dot). The shell integration already
    // sets PXH_HOSTNAME via `hostname -s`, but when falling back to
    // hostname::get() the result may be an FQDN, so strip the domain suffix.
    let hostname_bytes = hostname.as_bytes();
    if let Some(dot_pos) = hostname_bytes.iter().position(|&b| b == b'.') {
        BString::from(&hostname_bytes[..dot_pos])
    } else {
        BString::from(hostname_bytes)
    }
}

/// Resolve symlinks in a path, even if the final target doesn't exist yet.
/// Walks components left-to-right: canonicalizes each prefix that exists on disk,
/// follows symlinks whose targets don't exist yet, and appends remaining components.
fn resolve_through_symlinks(path: &Path) -> PathBuf {
    let mut resolved = PathBuf::new();
    for component in path.components() {
        resolved.push(component);
        if let Ok(canonical) = std::fs::canonicalize(&resolved) {
            resolved = canonical;
        } else if let Ok(target) = std::fs::read_link(&resolved) {
            // Symlink exists but target doesn't -- follow it anyway
            if target.is_absolute() {
                resolved = target;
            } else {
                resolved.pop();
                resolved.push(target);
            }
        }
    }
    resolved
}

/// Resolve hostname: config > DB setting (legacy "original_hostname") > live hostname.
pub fn resolve_hostname(config: &recall::config::Config, conn: &Connection) -> BString {
    if let Some(ref h) = config.host.hostname {
        return BString::from(h.as_bytes());
    }
    get_setting(conn, "original_hostname").ok().flatten().unwrap_or_else(get_hostname)
}

/// Build the set of hostnames that count as "this host" for recall filtering.
/// Returns {current_hostname} ∪ {aliases}, deduped.
pub fn effective_host_set(config: &recall::config::Config) -> Vec<BString> {
    let current = get_hostname();
    let mut hosts = vec![current];
    for alias in &config.host.aliases {
        let b = BString::from(alias.as_bytes());
        if !hosts.contains(&b) {
            hosts.push(b);
        }
    }
    hosts
}

/// Migrate host settings from DB legacy storage to config file.
/// Called from install and config commands, not on every connection.
///
/// - If config lacks hostname, read from DB (legacy "original_hostname"), then delete from DB.
/// - If config hostname doesn't match live hostname, move old to aliases and update.
/// - If config lacks machine_id, generate one.
pub fn migrate_host_settings(conn: &Connection) {
    if recall::config::Config::has_parse_error() {
        log::warn!("config.toml has parse errors; skipping host-settings migration");
        return;
    }
    let config = recall::config::Config::load();
    let mut updates: Vec<(&str, toml_edit::Item)> = Vec::new();
    let live_hostname = get_hostname();

    // Step 1: Establish config hostname (from DB legacy or live)
    let config_hostname = if let Some(ref h) = config.host.hostname {
        BString::from(h.as_bytes())
    } else if let Ok(Some(hostname)) = get_setting(conn, "original_hostname") {
        updates.push(("host.hostname", toml_edit::value(hostname.to_string())));
        hostname
    } else {
        updates.push(("host.hostname", toml_edit::value(live_hostname.to_string())));
        live_hostname.clone()
    };

    // Step 2: If hostname changed, move old to aliases and update
    if config_hostname != live_hostname {
        let mut aliases = config.host.aliases.clone();
        let old_str = config_hostname.to_string();
        if !aliases.contains(&old_str) {
            aliases.push(old_str);
        }
        let alias_array = toml_edit::Array::from_iter(aliases.iter().map(|s| s.as_str()));
        updates.push(("host.aliases", toml_edit::value(alias_array)));
        updates.push(("host.hostname", toml_edit::value(live_hostname.to_string())));
    }

    // Step 3: Generate machine_id if missing
    if config.host.machine_id.is_none() {
        let id = rand::random::<u64>();
        // Mask to 63 bits so the value stays positive in TOML (signed i64).
        // A negative i64 can't deserialize into Option<u64>, which would cause
        // the entire config file to fail to parse.
        let id = id & i64::MAX as u64;
        updates.push(("host.machine_id", toml_edit::value(id as i64)));
    }

    if !updates.is_empty()
        && let Err(e) = recall::config::Config::update_default_config(&updates)
    {
        log::warn!("Failed to migrate host settings to config: {e}");
        return;
    }

    // Clear legacy original_hostname from DB after successful config write
    if config.host.hostname.is_none() {
        let _ = conn.execute("DELETE FROM settings WHERE key = 'original_hostname'", []);
    }

    // Mirror machine_id into the DB so sync can identify this database by its
    // machine identity (independent of file path). Re-load config to pick up
    // any value we just wrote above.
    if let Some(machine_id) = recall::config::Config::load().host.machine_id {
        let bs = BString::from(machine_id.to_string());
        let _ = set_setting(conn, "local_machine_id", &bs);
    }
}
/// Return the pxh data directory. Prefers XDG, falls back to `~/.pxh` if it exists.
/// New installs default to `$XDG_DATA_HOME/pxh` (`~/.local/share/pxh`).
pub fn pxh_data_dir() -> Option<PathBuf> {
    let home = home::home_dir()?;
    let xdg_data =
        env::var("XDG_DATA_HOME").map(PathBuf::from).unwrap_or_else(|_| home.join(".local/share"));
    let xdg_dir = xdg_data.join("pxh");
    if xdg_dir.exists() {
        return Some(xdg_dir);
    }
    let legacy = home.join(".pxh");
    if legacy.exists() {
        return Some(legacy);
    }
    Some(xdg_dir)
}

/// Return the pxh config directory. Prefers XDG, falls back to `~/.pxh` if it exists.
/// New installs default to `$XDG_CONFIG_HOME/pxh` (`~/.config/pxh`).
pub fn pxh_config_dir() -> Option<PathBuf> {
    let home = home::home_dir()?;
    let xdg_config =
        env::var("XDG_CONFIG_HOME").map(PathBuf::from).unwrap_or_else(|_| home.join(".config"));
    let xdg_dir = xdg_config.join("pxh");
    if xdg_dir.exists() {
        return Some(xdg_dir);
    }
    let legacy = home.join(".pxh");
    if legacy.exists() {
        return Some(legacy);
    }
    Some(xdg_dir)
}

/// Return the default database path (`pxh_data_dir()/pxh.db`).
pub fn default_db_path() -> Option<PathBuf> {
    Some(pxh_data_dir()?.join("pxh.db"))
}

/// Initialize base schema (persistent tables and indexes only).
/// Safe to call on foreign databases (scan/scrub --dir) -- all DDL is idempotent.
pub fn initialize_base_schema(conn: &Connection) -> Result<(), Box<dyn std::error::Error>> {
    conn.execute_batch(include_str!("base_schema.sql"))?;
    Ok(())
}

/// Set up the in-memory memdb tables and register the REGEXP function.
/// Only needed by commands that use `memdb.show_results` or REGEXP (show, scrub).
pub fn initialize_full_schema(conn: &Connection) -> Result<(), Box<dyn std::error::Error>> {
    conn.execute_batch(
        "ATTACH DATABASE ':memory:' AS memdb;
         CREATE TABLE memdb.show_results (
             ch_rowid INTEGER NOT NULL,
             ch_start_unix_timestamp INTEGER,
             ch_id INTEGER NOT NULL
         );
         CREATE INDEX memdb.result_timestamp ON show_results(ch_start_unix_timestamp, ch_id);",
    )?;
    conn.create_scalar_function("regexp", 2, FunctionFlags::SQLITE_DETERMINISTIC, move |ctx| {
        assert_eq!(ctx.len(), 2, "called with unexpected number of arguments");
        let regexp: Arc<Regex> = ctx
            .get_or_create_aux(0, |vr| -> Result<_, BoxError> { Ok(Regex::new(vr.as_str()?)?) })?;
        let is_match = {
            let text = ctx.get_raw(1).as_bytes().map_err(|e| Error::UserFunctionError(e.into()))?;
            regexp.is_match(text)
        };
        Ok(is_match)
    })?;
    Ok(())
}

/// Current schema version -- bump when adding new migrations below.
pub const CURRENT_SCHEMA_VERSION: i32 = 3;

/// Run versioned schema migrations tracked via PRAGMA user_version.
pub fn run_schema_migrations(conn: &Connection) -> Result<(), Box<dyn std::error::Error>> {
    let version: i32 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;

    if version < 1 {
        // Column may already exist on databases created before version tracking.
        match conn.execute("ALTER TABLE command_history ADD COLUMN machine_id INTEGER", []) {
            Ok(_) => {}
            Err(e) if e.to_string().contains("duplicate column name") => {}
            Err(e) => return Err(e.into()),
        }
        conn.pragma_update(None, "user_version", 1)?;
    }

    if version < 2 {
        // Composite index for fast MAX(id) lookups in seal's UPDATE query.
        conn.execute_batch(
            "CREATE INDEX IF NOT EXISTS idx_session_id_desc ON command_history(session_id, id DESC)",
        )?;
        conn.pragma_update(None, "user_version", 2)?;
    }

    if version < 3 {
        // v3 introduces settings.local_machine_id so sync can identify a DB by
        // its machine identity rather than by file path. The actual write
        // happens in migrate_host_settings (Install/Config of the local DB
        // only) -- doing it here would contaminate foreign databases that
        // also pass through run_schema_migrations during the sync merge.
        conn.pragma_update(None, "user_version", 3)?;
    }

    Ok(())
}

/// Open a lightweight connection (persistent schema only, no memdb/regexp).
/// Use for hot-path commands like insert and seal.
pub fn sqlite_connection(path: &Option<PathBuf>) -> Result<Connection, Box<dyn std::error::Error>> {
    sqlite_connection_inner(path, false)
}

/// Open a full connection with memdb tables and REGEXP function.
/// Use for commands that need `memdb.show_results` or REGEXP (show, scrub, scan).
pub fn sqlite_connection_full(
    path: &Option<PathBuf>,
) -> Result<Connection, Box<dyn std::error::Error>> {
    sqlite_connection_inner(path, true)
}

fn sqlite_connection_inner(
    path: &Option<PathBuf>,
    full: bool,
) -> Result<Connection, Box<dyn std::error::Error>> {
    let path = path.as_ref().ok_or("Database not defined; use --db or PXH_DB_PATH")?;
    if let Some(parent) = path.parent() {
        // Follow symlinks so create_dir_all creates the real target directory
        // rather than conflicting with an existing symlink entry.
        let resolved = resolve_through_symlinks(parent);
        std::fs::create_dir_all(resolved)?;
    }
    let conn = Connection::open(path)?;

    // Ensure the database file is only readable by the owner
    use std::os::unix::fs::PermissionsExt;
    if let Ok(metadata) = std::fs::metadata(path) {
        let mode = metadata.permissions().mode();
        if mode & 0o077 != 0 {
            std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?;
        }
    }

    conn.busy_timeout(Duration::from_millis(5000))?;
    conn.pragma_update(None, "journal_mode", "WAL")?;
    conn.pragma_update(None, "temp_store", "MEMORY")?;
    conn.pragma_update(None, "cache_size", -65536_i64)?; // 64 MB (negative = KiB)
    conn.pragma_update(None, "mmap_size", 268435456_i64)?; // 256 MB
    conn.pragma_update(None, "synchronous", "NORMAL")?;

    // Schema DDL is idempotent but not free -- skip it once the database is at
    // the current version. Hot-path commands (insert, seal) open a fresh
    // connection per invocation, so this saves a parse + several PRAGMA calls
    // every time.
    let version: i32 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
    if version != CURRENT_SCHEMA_VERSION {
        initialize_base_schema(&conn)?;
        run_schema_migrations(&conn)?;
    }

    if full {
        initialize_full_schema(&conn)?;
    }

    Ok(conn)
}

/// Run `f` against a write transaction, retrying on `SQLITE_BUSY`/`SQLITE_LOCKED`
/// with exponential backoff and jitter. Each attempt uses `BEGIN IMMEDIATE` so
/// contention is detected up front rather than mid-transaction. Gives up once
/// `max_total` elapses and returns the final busy error.
///
/// Why this exists: SQLite's own `busy_timeout` retries are unfair under heavy
/// write contention (a single waiter can spend the full timeout sleeping while
/// other writers slip past). Jittered backoff at the caller gives fairness and
/// a bounded worst-case latency suitable for the shell hot path.
pub fn with_write_retry<T, F>(
    conn: &mut Connection,
    max_total: Duration,
    mut f: F,
) -> Result<T, Box<dyn std::error::Error>>
where
    F: FnMut(&Transaction) -> rusqlite::Result<T>,
{
    let start = Instant::now();
    let mut backoff = Duration::from_micros(500);
    let cap = Duration::from_millis(50);
    loop {
        let attempt = (|| -> rusqlite::Result<T> {
            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
            let value = f(&tx)?;
            tx.commit()?;
            Ok(value)
        })();
        match attempt {
            Ok(v) => return Ok(v),
            Err(e) if is_busy(&e) => {
                let elapsed = start.elapsed();
                if elapsed >= max_total {
                    return Err(e.into());
                }
                let jitter_us = rand::random::<u64>() % backoff.as_micros().max(1) as u64;
                let sleep = backoff + Duration::from_micros(jitter_us);
                std::thread::sleep(sleep.min(max_total - elapsed));
                backoff = (backoff * 2).min(cap);
            }
            Err(e) => return Err(e.into()),
        }
    }
}

fn is_busy(e: &rusqlite::Error) -> bool {
    matches!(
        e,
        rusqlite::Error::SqliteFailure(err, _)
            if matches!(err.code, ErrorCode::DatabaseBusy | ErrorCode::DatabaseLocked)
    )
}

/// Read this database's `local_machine_id` setting, parsed as u64.
/// Accepts either TEXT or BLOB storage so we tolerate values written by
/// non-pxh tooling. Returns None if the setting is absent or unparseable --
/// callers should fall back to non-watermarked behavior in that case.
pub fn read_local_machine_id(conn: &Connection) -> Option<u64> {
    conn.query_row("SELECT value FROM settings WHERE key = 'local_machine_id'", [], |row| {
        let bytes = row.get_ref(0)?.as_bytes().unwrap_or(&[]);
        Ok(std::str::from_utf8(bytes).ok().and_then(|s| s.parse::<u64>().ok()))
    })
    .ok()
    .flatten()
}

#[derive(Debug, Default, Serialize, Deserialize)]
pub struct Invocation {
    pub command: BString,
    pub shellname: String,
    pub working_directory: Option<BString>,
    pub hostname: Option<BString>,
    pub username: Option<BString>,
    pub exit_status: Option<i64>,
    pub start_unix_timestamp: Option<i64>,
    pub end_unix_timestamp: Option<i64>,
    pub session_id: i64,
    #[serde(default)]
    pub machine_id: Option<u64>,
}

impl Invocation {
    fn sameish(&self, other: &Self) -> bool {
        self.command == other.command && self.start_unix_timestamp == other.start_unix_timestamp
    }

    pub fn insert(&self, tx: &Transaction) -> rusqlite::Result<()> {
        tx.execute(
            r#"
INSERT OR IGNORE INTO command_history (
    session_id,
    full_command,
    shellname,
    hostname,
    username,
    working_directory,
    exit_status,
    start_unix_timestamp,
    end_unix_timestamp,
    machine_id
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"#,
            (
                self.session_id,
                self.command.as_slice(),
                self.shellname.clone(),
                self.hostname.as_ref().map(|v| v.to_vec()),
                self.username.as_ref().map(|v| v.to_vec()),
                self.working_directory.as_ref().map(|v| v.to_vec()),
                self.exit_status,
                self.start_unix_timestamp,
                self.end_unix_timestamp,
                self.machine_id.map(|id| id as i64),
            ),
        )?;
        Ok(())
    }
}

// Try to generate a "stable" session id based on the file imported.
// If that fails, just create a random one.
fn generate_import_session_id(histfile: &Path) -> i64 {
    if let Ok(st) = std::fs::metadata(histfile) {
        ((st.ino() << 16) | st.dev()) as i64
    } else {
        (rand::random::<u64>() >> 1) as i64
    }
}

/// Join backslash-continuation lines in zsh EXTENDED_HISTORY format.
/// Lines ending with `\` are joined with their successor using `\n` as separator.
pub fn join_continuation_lines(buf: &[u8]) -> Vec<Vec<u8>> {
    let mut result: Vec<Vec<u8>> = Vec::new();
    let mut current: Vec<u8> = Vec::new();
    for line in buf.split(|&ch| ch == b'\n') {
        if line.last() == Some(&b'\\') {
            current.extend_from_slice(&line[..line.len() - 1]);
            current.push(b'\n');
        } else {
            current.extend_from_slice(line);
            if !current.is_empty() {
                result.push(std::mem::take(&mut current));
            }
        }
    }
    if !current.is_empty() {
        result.push(current);
    }
    result
}

pub fn import_zsh_history(
    histfile: &Path,
    hostname: Option<BString>,
    username: Option<BString>,
) -> Result<Vec<Invocation>, Box<dyn std::error::Error>> {
    let mut f = File::open(histfile)?;
    let mut buf = Vec::new();
    let _ = f.read_to_end(&mut buf)?;
    let username = username
        .or_else(|| uzers::get_current_username().map(|v| BString::from(v.into_vec())))
        .unwrap_or_else(|| BString::from("unknown"));
    let hostname = hostname.unwrap_or_else(get_hostname);
    // Pre-join backslash-continuation lines: zsh EXTENDED_HISTORY writes
    // multi-line commands as physical lines ending with '\'.
    let logical_lines = join_continuation_lines(&buf);

    let mut ret = vec![];
    let mut skipped = 0usize;
    let session_id = generate_import_session_id(histfile);
    for (line_num, line) in logical_lines.iter().enumerate() {
        let line = line.as_slice();
        let Some((fields, command)) = line.splitn(2, |&ch| ch == b';').collect_tuple() else {
            continue;
        };
        let Some((_skip, start_time, duration_seconds)) =
            fields.splitn(3, |&ch| ch == b':').collect_tuple()
        else {
            continue;
        };
        let start_unix_timestamp = match str::from_utf8(start_time.get(1..).unwrap_or(&[]))
            .ok()
            .and_then(|s| s.parse::<i64>().ok())
        {
            Some(ts) => ts,
            None => {
                eprintln!(
                    "warning: {}: skipping line {}: bad timestamp {:?}",
                    histfile.display(),
                    line_num + 1,
                    BString::from(start_time),
                );
                skipped += 1;
                continue;
            }
        };
        let duration =
            match str::from_utf8(duration_seconds).ok().and_then(|s| s.parse::<i64>().ok()) {
                Some(d) => d,
                None => {
                    eprintln!(
                        "warning: {}: skipping line {}: bad duration {:?}",
                        histfile.display(),
                        line_num + 1,
                        BString::from(duration_seconds),
                    );
                    skipped += 1;
                    continue;
                }
            };
        let invocation = Invocation {
            command: BString::from(command),
            shellname: "zsh".into(),
            hostname: Some(BString::from(hostname.as_bytes())),
            username: Some(BString::from(username.as_bytes())),
            start_unix_timestamp: Some(start_unix_timestamp),
            end_unix_timestamp: Some(start_unix_timestamp + duration),
            session_id,
            ..Default::default()
        };

        ret.push(invocation);
    }

    if skipped > 0 {
        eprintln!("warning: {}: skipped {skipped} malformed line(s)", histfile.display());
    }

    Ok(dedup_invocations(ret))
}

pub fn import_bash_history(
    histfile: &Path,
    hostname: Option<BString>,
    username: Option<BString>,
) -> Result<Vec<Invocation>, Box<dyn std::error::Error>> {
    let mut f = File::open(histfile)?;
    let mut buf = Vec::new();
    let _ = f.read_to_end(&mut buf)?;
    let username = username
        .or_else(|| uzers::get_current_username().map(|v| BString::from(v.as_bytes())))
        .unwrap_or_else(|| BString::from("unknown"));
    let hostname = hostname.unwrap_or_else(get_hostname);
    let buf_iter = buf.split(|&ch| ch == b'\n').filter(|l| !l.is_empty());

    let mut ret = vec![];
    let session_id = generate_import_session_id(histfile);
    let mut last_ts = None;
    for line in buf_iter {
        if line[0] == b'#'
            && let Ok(ts) = str::parse::<i64>(str::from_utf8(&line[1..]).unwrap_or(""))
        {
            if ts > 0 {
                last_ts = Some(ts);
            }
            continue;
        }
        let invocation = Invocation {
            command: BString::from(line),
            shellname: "bash".into(),
            hostname: Some(BString::from(hostname.as_bytes())),
            username: Some(BString::from(username.as_bytes())),
            start_unix_timestamp: last_ts,
            session_id,
            ..Default::default()
        };

        ret.push(invocation);
    }

    Ok(dedup_invocations(ret))
}

pub fn import_json_history(histfile: &Path) -> Result<Vec<Invocation>, Box<dyn std::error::Error>> {
    let f = File::open(histfile)?;
    let reader = BufReader::new(f);
    Ok(serde_json::from_reader(reader)?)
}

fn dedup_invocations(invocations: Vec<Invocation>) -> Vec<Invocation> {
    let mut it = invocations.into_iter();
    let Some(first) = it.next() else { return vec![] };
    let mut ret = vec![first];
    for elem in it {
        if !elem.sameish(ret.last().unwrap()) {
            ret.push(elem);
        }
    }
    ret
}

impl Invocation {
    pub fn from_row(row: &Row) -> Result<Self, Error> {
        Ok(Invocation {
            session_id: row.get("session_id")?,
            command: BString::from(row.get::<_, Vec<u8>>("full_command")?),
            shellname: row.get("shellname")?,
            working_directory: row
                .get::<_, Option<Vec<u8>>>("working_directory")?
                .map(BString::from),
            hostname: row.get::<_, Option<Vec<u8>>>("hostname")?.map(BString::from),
            username: row.get::<_, Option<Vec<u8>>>("username")?.map(BString::from),
            exit_status: row.get("exit_status")?,
            start_unix_timestamp: row.get("start_unix_timestamp")?,
            end_unix_timestamp: row.get("end_unix_timestamp")?,
            machine_id: row.get::<_, Option<i64>>("machine_id").ok().flatten().map(|v| v as u64),
        })
    }
}

// Create a pretty export string that gets serialized as an array of
// bytes only if it isn't valid UTF-8; this makes the json export
// prettier.
#[derive(Debug, Eq, PartialEq, Serialize, Deserialize, Clone)]
#[serde(untagged)]
enum PrettyExportString {
    Readable(String),
    Encoded(Vec<u8>),
}

impl From<&[u8]> for PrettyExportString {
    fn from(bytes: &[u8]) -> Self {
        match str::from_utf8(bytes) {
            Ok(v) => Self::Readable(v.to_string()),
            _ => Self::Encoded(bytes.to_vec()),
        }
    }
}

impl From<Option<&Vec<u8>>> for PrettyExportString {
    fn from(bytes: Option<&Vec<u8>>) -> Self {
        match bytes {
            Some(v) => match str::from_utf8(v.as_slice()) {
                Ok(s) => Self::Readable(s.to_string()),
                _ => Self::Encoded(v.to_vec()),
            },
            None => Self::Readable(String::new()),
        }
    }
}

impl Invocation {
    fn to_json_export(&self) -> serde_json::Value {
        serde_json::json!({
            "session_id": self.session_id,
            "command": PrettyExportString::from(self.command.as_slice()),
            "shellname": self.shellname,
            "working_directory": self.working_directory.as_ref().map(
                |b| PrettyExportString::from(b.as_slice())
            ),
            "hostname": self.hostname.as_ref().map(
                |b| PrettyExportString::from(b.as_slice())
            ),
            "username": self.username.as_ref().map(
                |b| PrettyExportString::from(b.as_slice())
            ),
            "exit_status": self.exit_status,
            "start_unix_timestamp": self.start_unix_timestamp,
            "end_unix_timestamp": self.end_unix_timestamp,
            "machine_id": self.machine_id,
        })
    }
}

pub fn json_export(rows: &[Invocation]) -> Result<(), Box<dyn std::error::Error>> {
    let json_values: Vec<serde_json::Value> = rows.iter().map(|r| r.to_json_export()).collect();
    serde_json::to_writer(io::stdout(), &json_values)?;
    Ok(())
}

// column list: command, start, host, shell, cwd, end, duratio, session, ...

struct QueryResultColumnDisplayer {
    header: &'static str,
    header_style: &'static str,
    displayer: Box<dyn Fn(&Invocation) -> prettytable::Cell>,
}

fn time_display_helper(t: Option<i64>) -> String {
    // Chained if-let may make this unpacking of
    // Option/Result/LocalResult cleaner.  Alternative is a closer
    // using `?` chains but that's slightly uglier.
    t.and_then(|t| Local.timestamp_opt(t, 0).single())
        .map(|t| t.format(TIME_FORMAT).to_string())
        .unwrap_or_else(|| "n/a".to_string())
}

fn binary_display_helper(v: &BString) -> String {
    String::from_utf8_lossy(v.as_slice()).to_string()
}

fn displayers() -> HashMap<&'static str, QueryResultColumnDisplayer> {
    let mut ret = HashMap::new();
    ret.insert(
        "command",
        QueryResultColumnDisplayer {
            header: "Command",
            header_style: "Fw",
            displayer: Box::new(|row| {
                prettytable::Cell::new(&binary_display_helper(&row.command)).style_spec("Fw")
            }),
        },
    );
    ret.insert(
        "start_time",
        QueryResultColumnDisplayer {
            header: "Start",
            header_style: "Fg",
            displayer: Box::new(|row| {
                prettytable::Cell::new(&time_display_helper(row.start_unix_timestamp))
                    .style_spec("Fg")
            }),
        },
    );
    ret.insert(
        "end_time",
        QueryResultColumnDisplayer {
            header: "End",
            header_style: "Fg",
            displayer: Box::new(|row| {
                prettytable::Cell::new(&time_display_helper(row.end_unix_timestamp))
                    .style_spec("Fg")
            }),
        },
    );
    ret.insert(
        "duration",
        QueryResultColumnDisplayer {
            header: "Duration",
            header_style: "Fm",
            displayer: Box::new(|row| {
                let text = match (row.start_unix_timestamp, row.end_unix_timestamp) {
                    (Some(start), Some(end)) => format!("{}s", end - start),
                    _ => "n/a".into(),
                };
                prettytable::Cell::new(&text).style_spec("Fm")
            }),
        },
    );
    ret.insert(
        "status",
        QueryResultColumnDisplayer {
            header: "Status",
            header_style: "Fr",
            displayer: Box::new(|row| match row.exit_status {
                Some(0) => prettytable::Cell::new("0").style_spec("Fg"),
                Some(s) => prettytable::Cell::new(&s.to_string()).style_spec("Fr"),
                None => prettytable::Cell::new("n/a").style_spec("Fd"),
            }),
        },
    );
    // TODO: Make session similar to "context" and just print `.` when
    // it is the current session.
    ret.insert(
        "session",
        QueryResultColumnDisplayer {
            header: "Session",
            header_style: "Fc",
            displayer: Box::new(|row| {
                prettytable::Cell::new(&format!("{}", row.session_id)).style_spec("Fc")
            }),
        },
    );
    // Print context specially; the full output is $HOST:$PATH but if
    // $HOST is the current host, the $HOST: is omitted.  If $PATH is
    // the current working directory, it is replaced with `.`.
    ret.insert(
        "context",
        QueryResultColumnDisplayer {
            header: "Context",
            header_style: "bFb",
            displayer: Box::new(|row| {
                let current_hostname = get_hostname();
                let row_hostname = row.hostname.clone().unwrap_or_default();
                let mut ret = String::new();
                if current_hostname != row_hostname {
                    write!(ret, "{row_hostname}:").unwrap_or_default();
                }
                let current_directory = env::current_dir().unwrap_or_default();
                ret.push_str(&row.working_directory.as_ref().map_or_else(String::new, |v| {
                    let v = String::from_utf8_lossy(v.as_slice()).to_string();
                    if v == current_directory.to_string_lossy() { String::from(".") } else { v }
                }));

                prettytable::Cell::new(&ret).style_spec("bFb")
            }),
        },
    );

    ret
}

pub fn present_results_human_readable(
    fields: &[&str],
    rows: &[Invocation],
    suppress_headers: bool,
) -> Result<(), Box<dyn std::error::Error>> {
    let displayers = displayers();
    let mut table = prettytable::Table::new();
    table.set_format(*prettytable::format::consts::FORMAT_CLEAN);

    if !suppress_headers {
        let mut title_row = prettytable::Row::empty();
        for field in fields {
            let Some(d) = displayers.get(field) else {
                return Err(Box::from(format!("Invalid 'show' field: {field}")));
            };

            title_row.add_cell(prettytable::Cell::new(d.header).style_spec(d.header_style));
        }
        table.set_titles(title_row);
    }

    for row in rows.iter() {
        let is_failed = matches!(row.exit_status, Some(s) if s != 0);
        let mut display_row = prettytable::Row::empty();
        for field in fields {
            let cell = (displayers[field].displayer)(row);
            if is_failed {
                display_row.add_cell(prettytable::Cell::new(&cell.get_content()).style_spec("Fr"));
            } else {
                display_row.add_cell(cell);
            }
        }
        table.add_row(display_row);
    }
    table.printstd();
    Ok(())
}

// Rewrite a file with lines matching `contraband` removed.  utf-8
// safe for the file (TODO: I guess make contraband a `BString` too)
pub fn atomically_remove_lines_from_file(
    input_filepath: &Path,
    contraband: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    // Resolve symlinks so persist() replaces the real target, not the symlink
    let resolved = resolve_through_symlinks(input_filepath);
    let original_perms = std::fs::metadata(&resolved)?.permissions();
    let input_file = File::open(&resolved)?;
    let mut input_reader = BufReader::new(input_file);

    let parent = resolved.parent().unwrap_or(Path::new("."));
    let temp_file = tempfile::NamedTempFile::new_in(parent)?;
    let mut output_writer = BufWriter::new(&temp_file);

    input_reader.for_byte_line_with_terminator(|line| {
        if !line.contains_str(contraband) {
            output_writer.write_all(line)?;
        }
        Ok(true)
    })?;

    output_writer.flush()?;
    drop(output_writer);
    temp_file.persist(&resolved)?;
    std::fs::set_permissions(&resolved, original_perms)?;
    Ok(())
}

/// Rewrite a file removing lines that exactly match any of the contraband items.
/// Unlike `atomically_remove_lines_from_file`, this matches complete lines (after trimming).
pub fn atomically_remove_matching_lines_from_file(
    input_filepath: &Path,
    contraband_items: &[&str],
) -> Result<(), Box<dyn std::error::Error>> {
    use std::collections::HashSet;

    // Resolve symlinks so persist() replaces the real target, not the symlink
    let resolved = resolve_through_symlinks(input_filepath);
    let original_perms = std::fs::metadata(&resolved)?.permissions();
    let contraband_set: HashSet<&str> = contraband_items.iter().copied().collect();

    let input_file = File::open(&resolved)?;
    let mut input_reader = BufReader::new(input_file);

    let parent = resolved.parent().unwrap_or(Path::new("."));
    let temp_file = tempfile::NamedTempFile::new_in(parent)?;
    let mut output_writer = BufWriter::new(&temp_file);

    input_reader.for_byte_line_with_terminator(|line| {
        let line_str = line.to_str_lossy();
        let trimmed = line_str.trim();
        if !contraband_set.contains(trimmed) {
            output_writer.write_all(line)?;
        }
        Ok(true)
    })?;

    output_writer.flush()?;
    drop(output_writer);
    temp_file.persist(&resolved)?;
    std::fs::set_permissions(&resolved, original_perms)?;
    Ok(())
}

// Helper functions for command parsing and path resolution
pub mod helpers {
    use std::ffi::OsString;
    use std::path::{Path, PathBuf};

    /// Parse an SSH command string into command and arguments, handling quotes and spaces.
    /// Similar to how rsync and other tools parse the -e option.
    pub fn parse_ssh_command(ssh_cmd: &str) -> (String, Vec<String>) {
        // If it's a simple command without spaces, just return it
        if !ssh_cmd.contains(char::is_whitespace) {
            return (ssh_cmd.to_string(), vec![]);
        }

        // Otherwise, we need to parse it properly
        let mut cmd = String::new();
        let mut args = Vec::new();
        let mut current = String::new();
        let mut in_quotes = false;
        let mut quote_char = '\0';
        let mut is_first = true;
        let mut chars = ssh_cmd.chars().peekable();

        while let Some(ch) = chars.next() {
            match ch {
                '"' | '\'' if !in_quotes => {
                    in_quotes = true;
                    quote_char = ch;
                }
                '"' | '\'' if in_quotes && ch == quote_char => {
                    in_quotes = false;
                    quote_char = '\0';
                }
                ' ' | '\t' if !in_quotes => {
                    if !current.is_empty() {
                        if is_first {
                            cmd = current.clone();
                            is_first = false;
                        } else {
                            args.push(current.clone());
                        }
                        current.clear();
                    }
                }
                '\\' if chars.peek().is_some() => {
                    // Handle escaped characters
                    if let Some(next_ch) = chars.next() {
                        current.push(next_ch);
                    }
                }
                _ => {
                    current.push(ch);
                }
            }
        }

        // Don't forget the last token
        if !current.is_empty() {
            if is_first {
                cmd = current;
            } else {
                args.push(current);
            }
        }

        (cmd, args)
    }

    /// Build a list of candidate paths to search for pxh on the remote host.
    /// The first candidate that exists and is executable will be used.
    fn remote_pxh_candidates(configured_path: &str) -> Vec<String> {
        let mut candidates = Vec::new();

        if configured_path != "pxh" {
            candidates.push(configured_path.to_string());
            return candidates;
        }

        // Try the same relative-to-home path as the local binary
        if let Some(rel) = get_relative_path_from_home(None, None)
            && rel != "pxh"
        {
            candidates.push(format!("$HOME/{rel}"));
        }

        // Common installation locations
        for p in [
            "$HOME/.cargo/bin/pxh",
            "$HOME/bin/pxh",
            "$HOME/.local/bin/pxh",
            "/usr/local/bin/pxh",
            "/usr/bin/pxh",
        ] {
            if !candidates.contains(&p.to_string()) {
                candidates.push(p.to_string());
            }
        }

        candidates
    }

    /// Build a shell command that finds and executes pxh on the remote host.
    /// When a single explicit path is configured, uses it directly.
    /// Otherwise, probes a prioritized list of candidate locations.
    pub fn build_remote_pxh_command(configured_path: &str, args: &str) -> String {
        let candidates = remote_pxh_candidates(configured_path);

        if candidates.len() == 1 {
            return format!("{} {args}", candidates[0]);
        }

        // Build a shell snippet that tries each candidate in order
        let checks: Vec<String> =
            candidates.iter().map(|p| format!("[ -x \"{p}\" ] && exec \"{p}\" {args}")).collect();
        format!(
            "sh -c '{}; echo \"pxh: not found on remote host\" >&2; exit 127'",
            checks.join("; ")
        )
    }

    /// Gets the relative path from home directory if the current executable is within it.
    /// Returns None if the executable is not in the home directory.
    /// Takes optional overrides for testing.
    pub fn get_relative_path_from_home(
        exe_override: Option<&Path>,
        home_override: Option<&Path>,
    ) -> Option<String> {
        let exe = match exe_override {
            Some(path) => path.to_path_buf(),
            None => std::env::current_exe().ok()?,
        };

        let home = match home_override {
            Some(path) => path.to_path_buf(),
            None => home::home_dir()?,
        };

        exe.strip_prefix(&home).ok().map(|path| path.to_string_lossy().to_string())
    }

    /// Return a shell expression that resolves the pxh database path on a remote host.
    /// Checks XDG path first, falls back to legacy ~/.pxh.
    pub fn default_remote_db_expr() -> String {
        r#"$(if [ -d "${XDG_DATA_HOME:-$HOME/.local/share}/pxh" ]; then echo "${XDG_DATA_HOME:-$HOME/.local/share}/pxh/pxh.db"; elif [ -d "$HOME/.pxh" ]; then echo "$HOME/.pxh/pxh.db"; else echo "${XDG_DATA_HOME:-$HOME/.local/share}/pxh/pxh.db"; fi)"#.to_string()
    }

    /// Determine if the executable is being invoked as pxhs (shorthand for pxh show)
    pub fn determine_is_pxhs(args: &[OsString]) -> bool {
        args.first()
            .and_then(|arg| {
                PathBuf::from(arg).file_name().map(|name| name.to_string_lossy().contains("pxhs"))
            })
            .unwrap_or(false)
    }
}

/// Test utilities - only intended for use in tests
#[doc(hidden)]
pub mod test_utils {
    use std::{
        env,
        path::{Path, PathBuf},
        process::Command,
    };

    use rand::{RngExt, distr::Alphanumeric};
    use tempfile::TempDir;

    pub fn pxh_path() -> PathBuf {
        let mut path = std::env::current_exe().unwrap();
        path.pop(); // Remove test binary name
        path.pop(); // Remove 'deps'
        path.push("pxh");
        assert!(path.exists(), "pxh binary not found at {:?}", path);
        path
    }

    fn generate_random_string(length: usize) -> String {
        rand::rng().sample_iter(&Alphanumeric).take(length).map(char::from).collect()
    }

    fn get_standard_path() -> String {
        // Use getconf to get the standard PATH
        Command::new("getconf")
            .arg("PATH")
            .output()
            .ok()
            .and_then(|output| {
                if output.status.success() { String::from_utf8(output.stdout).ok() } else { None }
            })
            .map(|s| s.trim().to_string())
            .unwrap_or_else(|| {
                // Fallback to a reasonable default if getconf fails
                "/usr/bin:/bin:/usr/sbin:/sbin".to_string()
            })
    }

    /// Unified test helper for invoking pxh with proper isolation and environment setup
    pub struct PxhTestHelper {
        _tmpdir: TempDir,
        pub hostname: String,
        pub username: String,
        home_dir: PathBuf,
        db_path: PathBuf,
    }

    impl PxhTestHelper {
        pub fn new() -> Self {
            let tmpdir = TempDir::new().unwrap();
            let home_dir = tmpdir.path().to_path_buf();
            let db_path = home_dir.join(".pxh/pxh.db");

            // Create .pxh dir and a config with empty ignore_patterns so tests
            // aren't affected by default ignore patterns filtering trivial commands.
            let pxh_dir = home_dir.join(".pxh");
            std::fs::create_dir_all(&pxh_dir).unwrap();
            std::fs::write(pxh_dir.join("config.toml"), "[history]\nignore_patterns = []\n")
                .unwrap();

            PxhTestHelper {
                _tmpdir: tmpdir,
                hostname: generate_random_string(12),
                username: "testuser".to_string(),
                home_dir,
                db_path,
            }
        }

        pub fn with_custom_db_path(mut self, db_path: impl AsRef<Path>) -> Self {
            self.db_path = self.home_dir.join(db_path);
            self
        }

        /// Get the temporary directory path
        pub fn home_dir(&self) -> &Path {
            &self.home_dir
        }

        /// Get the database path
        pub fn db_path(&self) -> &Path {
            &self.db_path
        }

        /// Get the full PATH including pxh binary directory
        pub fn get_full_path(&self) -> String {
            format!("{}:{}", pxh_path().parent().unwrap().display(), get_standard_path())
        }

        /// Create a pxh command with all environment properly set up
        pub fn command(&self) -> Command {
            let mut cmd = Command::new(pxh_path());

            // Clear environment to ensure isolation
            cmd.env_clear();

            // Set consistent test environment
            cmd.env("HOME", &self.home_dir);
            cmd.env("PXH_DB_PATH", &self.db_path);
            cmd.env("PXH_HOSTNAME", &self.hostname);
            cmd.env("USER", &self.username);
            cmd.env("PATH", self.get_full_path());

            // Propagate coverage environment variables if they exist
            if let Ok(profile_file) = env::var("LLVM_PROFILE_FILE") {
                cmd.env("LLVM_PROFILE_FILE", profile_file);
            }
            if let Ok(llvm_cov) = env::var("CARGO_LLVM_COV") {
                cmd.env("CARGO_LLVM_COV", llvm_cov);
            }

            cmd
        }

        /// Convenience method to create a command with arguments
        pub fn command_with_args(&self, args: &[&str]) -> Command {
            let mut cmd = self.command();
            cmd.args(args);
            cmd
        }

        /// Create a shell command (bash/zsh) with proper environment for interactive testing
        pub fn shell_command(&self, shell: &str) -> Command {
            let mut cmd = Command::new(shell);

            // Force interactive mode - use login shell to ensure rc files are loaded
            cmd.arg("-i");
            cmd.env_clear();

            // Set consistent environment variables
            cmd.env("HOME", &self.home_dir);
            cmd.env("PXH_DB_PATH", &self.db_path);
            cmd.env("PXH_HOSTNAME", &self.hostname);
            cmd.env("PATH", self.get_full_path());

            // Set a clean environment for testing
            cmd.env("USER", &self.username);
            cmd.env("SHELL", shell);

            // For bash to properly load rc files in a minimal environment
            cmd.env("BASH_ENV", self.home_dir.join(".bashrc"));

            cmd
        }
    }

    impl Default for PxhTestHelper {
        fn default() -> Self {
            Self::new()
        }
    }
}

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

    /// Create an in-memory database with full schema and migrations applied.
    fn test_connection() -> Connection {
        let conn = Connection::open_in_memory().unwrap();
        initialize_base_schema(&conn).unwrap();
        run_schema_migrations(&conn).unwrap();
        conn
    }

    #[test]
    fn test_resolve_hostname_from_config() {
        let conn = test_connection();

        let mut config = recall::config::Config::default();
        config.host.hostname = Some("from-config".to_string());
        set_setting(&conn, "original_hostname", &BString::from("from-db")).unwrap();

        let result = resolve_hostname(&config, &conn);
        assert_eq!(result, BString::from("from-config"));
    }

    #[test]
    fn test_resolve_hostname_from_db() {
        let conn = test_connection();

        let config = recall::config::Config::default();
        set_setting(&conn, "original_hostname", &BString::from("from-db")).unwrap();

        let result = resolve_hostname(&config, &conn);
        assert_eq!(result, BString::from("from-db"));
    }

    #[test]
    fn test_resolve_hostname_live_fallback() {
        let conn = test_connection();

        let config = recall::config::Config::default();
        let result = resolve_hostname(&config, &conn);
        assert_eq!(result, get_hostname());
    }

    #[test]
    fn test_effective_host_set_no_aliases() {
        let config = recall::config::Config::default();
        let hosts = effective_host_set(&config);
        assert_eq!(hosts, vec![get_hostname()]);
    }

    #[test]
    fn test_effective_host_set_with_aliases() {
        let mut config = recall::config::Config::default();
        config.host.aliases = vec!["old-host".to_string(), "other-host".to_string()];
        let hosts = effective_host_set(&config);
        assert_eq!(hosts.len(), 3);
        assert_eq!(hosts[0], get_hostname());
        assert_eq!(hosts[1], BString::from("old-host"));
        assert_eq!(hosts[2], BString::from("other-host"));
    }

    #[test]
    fn test_effective_host_set_dedup() {
        let mut config = recall::config::Config::default();
        let current = get_hostname().to_string();
        config.host.aliases = vec![current, "other".to_string()];
        let hosts = effective_host_set(&config);
        assert_eq!(hosts.len(), 2);
    }

    #[test]
    fn test_migration_fresh_database() {
        let conn = Connection::open_in_memory().unwrap();
        conn.execute_batch(include_str!("base_schema.sql")).unwrap();

        let version: i32 = conn.pragma_query_value(None, "user_version", |row| row.get(0)).unwrap();
        assert_eq!(version, 0);

        run_schema_migrations(&conn).unwrap();

        let version: i32 = conn.pragma_query_value(None, "user_version", |row| row.get(0)).unwrap();
        assert_eq!(version, CURRENT_SCHEMA_VERSION);

        // machine_id column should exist
        conn.execute(
            "INSERT INTO command_history (session_id, full_command, shellname, machine_id) VALUES (1, X'6C73', 'bash', 42)",
            [],
        )
        .unwrap();
    }

    #[test]
    fn test_migration_idempotent() {
        let conn = test_connection();
        // Run migrations again -- should be a no-op
        run_schema_migrations(&conn).unwrap();

        let version: i32 = conn.pragma_query_value(None, "user_version", |row| row.get(0)).unwrap();
        assert_eq!(version, CURRENT_SCHEMA_VERSION);
    }

    #[test]
    fn test_migration_legacy_database_with_machine_id() {
        // Simulate a database that was used with post-machine_id pxh but has no version tracking
        let conn = Connection::open_in_memory().unwrap();
        conn.execute_batch(include_str!("base_schema.sql")).unwrap();
        conn.execute("ALTER TABLE command_history ADD COLUMN machine_id INTEGER", []).unwrap();

        let version: i32 = conn.pragma_query_value(None, "user_version", |row| row.get(0)).unwrap();
        assert_eq!(version, 0);

        // Should handle the duplicate column gracefully and bump version
        run_schema_migrations(&conn).unwrap();

        let version: i32 = conn.pragma_query_value(None, "user_version", |row| row.get(0)).unwrap();
        assert_eq!(version, CURRENT_SCHEMA_VERSION);
    }

    #[test]
    fn test_with_write_retry_succeeds_immediately() {
        let mut conn = Connection::open_in_memory().unwrap();
        conn.execute_batch("CREATE TABLE t (v INTEGER);").unwrap();
        with_write_retry(&mut conn, Duration::from_secs(1), |tx| {
            tx.execute("INSERT INTO t VALUES (?)", [42])?;
            Ok(())
        })
        .unwrap();
        let v: i64 = conn.query_row("SELECT v FROM t", [], |r| r.get(0)).unwrap();
        assert_eq!(v, 42);
    }

    #[test]
    fn test_with_write_retry_propagates_non_busy_error() {
        let mut conn = Connection::open_in_memory().unwrap();
        let result = with_write_retry(&mut conn, Duration::from_secs(1), |tx| {
            tx.execute("SELECT * FROM does_not_exist", [])?;
            Ok(())
        });
        assert!(result.is_err());
    }

    #[test]
    fn test_with_write_retry_retries_then_succeeds() {
        let mut conn = Connection::open_in_memory().unwrap();
        conn.execute_batch("CREATE TABLE t (v INTEGER);").unwrap();
        let mut calls = 0;
        with_write_retry(&mut conn, Duration::from_secs(1), |tx| {
            calls += 1;
            if calls < 3 {
                // Synthesize a busy error to drive the retry loop.
                return Err(rusqlite::Error::SqliteFailure(
                    rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_BUSY),
                    None,
                ));
            }
            tx.execute("INSERT INTO t VALUES (?)", [1])?;
            Ok(())
        })
        .unwrap();
        assert_eq!(calls, 3);
        let n: i64 = conn.query_row("SELECT COUNT(*) FROM t", [], |r| r.get(0)).unwrap();
        assert_eq!(n, 1);
    }

    #[test]
    fn test_with_write_retry_gives_up_after_timeout() {
        let mut conn = Connection::open_in_memory().unwrap();
        let start = Instant::now();
        let result: Result<(), _> = with_write_retry(&mut conn, Duration::from_millis(20), |_tx| {
            Err(rusqlite::Error::SqliteFailure(
                rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_BUSY),
                None,
            ))
        });
        let elapsed = start.elapsed();
        assert!(result.is_err());
        // Should give up roughly at the cap, not run forever.
        assert!(elapsed < Duration::from_millis(200), "took {elapsed:?}");
    }
}