#[cfg(any(feature = "feature_capable", feature = "lite"))]
use std::borrow::Cow;
#[allow(unused_imports)]
use std::fmt::Write as _;
#[cfg(target_family = "unix")]
use std::os::unix::process::ExitStatusExt;
#[cfg(feature = "polars")]
use std::sync::Arc;
use std::{
    cmp::min,
    env, fs,
    fs::File,
    io::{BufRead, BufReader, BufWriter, Read, Write},
    path::{Path, PathBuf},
    process::Command,
    str,
    sync::OnceLock,
    time::{Duration, Instant, SystemTime},
};

use csv::ByteRecord;
use csv_index::RandomAccessSimple;
use docopt::Docopt;
use filetime::FileTime;
use human_panic::setup_panic;
use indicatif::{HumanCount, ProgressBar, ProgressDrawTarget, ProgressStyle};
use log::{info, log_enabled, warn};
#[cfg(feature = "polars")]
use polars::prelude::Schema;
use reqwest::Client;
use serde::de::DeserializeOwned;
#[cfg(any(feature = "feature_capable", feature = "lite"))]
use serde::de::{Deserialize, Deserializer, Error};
use sysinfo::System;
use zip::read::root_dir_common_filter;

#[cfg(feature = "polars")]
use crate::cmd::count::polars_count_input;
use crate::{
    CURRENT_COMMAND, CliError, CliResult,
    cmd::stats::{JsonTypes, STATSDATA_TYPES_MAP, StatsData},
    config,
    config::{
        Config, DEFAULT_RDR_BUFFER_CAPACITY, DEFAULT_WTR_BUFFER_CAPACITY, Delimiter, SpecialFormat,
        get_delim_by_extension, get_special_format,
    },
    select::SelectColumns,
};

#[macro_export]
macro_rules! regex_oncelock {
    ($re:literal $(,)?) => {{
        static RE: std::sync::OnceLock<regex::Regex> = std::sync::OnceLock::new();
        #[allow(clippy::regex_creation_in_loops)] // false positive as we use oncelock
        RE.get_or_init(|| regex::Regex::new($re).expect("Invalid regex"))
    }};
}

// leave at least 20% of the available memory free
const DEFAULT_FREEMEMORY_HEADROOM_PCT: u8 = 20;

// safety margin for memory-aware chunking
// (uses 80% of available memory to leave headroom for system operations & other processes)
pub const SAFETY_MARGIN: f64 = 0.8;

const DEFAULT_BATCH_SIZE: usize = 50_000;

const DEFAULT_STATSCACHE_MODE: &str = "auto";

static ROW_COUNT: OnceLock<Option<u64>> = OnceLock::new();

static JOBS_TO_USE: OnceLock<usize> = OnceLock::new();

pub static QUIET_FLAG: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);

// Set once at startup by `init_allocator_runtime()` when jemalloc's
// `background_thread` is successfully enabled (Linux; rejected on macOS). When
// true, jemalloc purges freed pages on background threads at no RSS cost, so the
// per-command page-retention lever (`retain_alloc_pages_for_aggregation`) is
// skipped to avoid paying its higher RSS for no additional benefit.
// Only read under the jemalloc-gated retain path / version string; harmless
// elsewhere.
#[allow(dead_code)]
pub static BACKGROUND_THREADS_ACTIVE: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

// only consumed by non-lite commands (describegpt, lens, python, luau)
#[cfg(not(feature = "lite"))]
pub static FILE_PATH_PREFIX: &str = "file:";

pub type ByteString = Vec<u8>;

#[allow(dead_code)]
#[derive(Clone, Copy, PartialEq, Eq)]
pub enum StatsMode {
    Schema,
    /// Like `Schema`, but additionally computes quartiles (q1/median/q3)
    /// and mode. Used by `qsv profile` so its descriptive-statistics
    /// projection (e.g. the Croissant annotations) can surface the full
    /// extended stat set on a fresh run without a pre-built `--everything`
    /// stats cache.
    ProfileSchema,
    Frequency,
    FrequencyForceStats,
    #[cfg(feature = "polars")]
    PolarsSchema,
    Outliers,
    None,
}

#[allow(dead_code)]
#[derive(serde::Deserialize, Clone)]
pub struct SchemaArgs {
    pub flag_enum_threshold:  u64,
    pub flag_ignore_case:     bool,
    pub flag_strict_dates:    bool,
    pub flag_strict_formats:  bool,
    pub flag_pattern_columns: SelectColumns,
    pub flag_dates_whitelist: String,
    pub flag_prefer_dmy:      bool,
    pub flag_force:           bool,
    pub flag_stdout:          bool,
    pub flag_jobs:            Option<usize>,
    pub flag_polars:          bool,
    pub flag_no_headers:      bool,
    pub flag_delimiter:       Option<Delimiter>,
    pub arg_input:            Option<String>,
    pub flag_memcheck:        bool,
    pub flag_output:          Option<String>,
}

#[inline]
pub fn num_cpus() -> usize {
    num_cpus::get()
}

static QSV_PATH: OnceLock<String> = OnceLock::new();

pub const CARGO_BIN_NAME: &str = env!("CARGO_BIN_NAME");
pub const CARGO_PKG_VERSION: &str = env!("CARGO_PKG_VERSION");

const TARGET: &str = match option_env!("TARGET") {
    Some(target) => target,
    None => "Unknown_target",
};
const QSV_KIND: &str = match option_env!("QSV_KIND") {
    Some(kind) => kind,
    None => "installed",
};

#[cfg(feature = "polars")]
const QSV_POLARS_REV: &str = match option_env!("QSV_POLARS_REV") {
    Some(rev) => rev,
    None => "",
};

// Add constant for whitespace visualization
// the whitespace markers as as defined in
// https://doc.rust-lang.org/reference/whitespace.html
const WHITESPACE_MARKERS: &[(char, &str)] = &[
    // common whitespace markers other than space
    ('\t', "《→》"), // tab
    ('\n', "《¶》"), // newline
    ('\r', "《⏎》"), // carriage return
    // more obscure whitespace markers
    ('\u{000B}', "《⋮》"), // vertical tab
    ('\u{000C}', "《␌》"), // form feed
    // note: the horizontal tab (U+0009) is the same code point as '\t' above,
    // so it is intentionally not repeated here (the '\t' entry already covers it).
    ('\u{0085}', "《␤》"), // next line
    ('\u{200E}', "《␎》"), // left-to-right mark
    ('\u{200F}', "《␏》"), // right-to-left mark
    ('\u{2028}', "《␊》"), // line separator
    ('\u{2029}', "《␍》"), // paragraph separator
    // additional common whitespace markers beyond
    // https://doc.rust-lang.org/reference/whitespace.html
    ('\u{00A0}', "《⍽》"),     // non-breaking space
    ('\u{2003}', "《emsp》"),  // em space
    ('\u{2007}', "《figsp》"), // figure space
    ('\u{200B}', "《zwsp》"),  // zero width space
];

pub fn reset_sigpipe() {
    cfg_select! {
        unix => unsafe {
            libc::signal(libc::SIGPIPE, libc::SIG_DFL);
        },
        _ => {},
    }
}

pub fn current_exe() -> CliResult<PathBuf> {
    let exe_path = std::env::current_exe()?;
    Ok(exe_path)
}

/// Visualizes whitespace characters in a string by replacing them with visible markers
///
/// This function takes a string and returns a new string where whitespace characters
/// are replaced with visible Unicode markers to make them easier to see.
///
/// # Arguments
///
/// * `s` - The input string to visualize whitespace in
///
/// # Returns
///
/// A new String with whitespace characters replaced by visible markers
///
/// # Behavior
///
/// - If the input string contains only spaces, each space is replaced with "《_》"
/// - For other whitespace characters (tab, newline, etc), uses markers defined in
///   `WHITESPACE_MARKERS`
/// - Non-whitespace characters are left unchanged
/// - For strings with mixed content, single spaces are preserved as-is
///
/// # Examples
///
/// ```
/// let s = "hello\tworld\n";
/// let vis = visualize_whitespace(s);
/// assert_eq!(vis, "hello《→》world《¶》");
///
/// let spaces = "   ";
/// let vis = visualize_whitespace(spaces);
/// assert_eq!(vis, "《_》《_》《_》");
/// ```
pub fn visualize_whitespace(s: &str) -> String {
    // Check if string is all spaces
    let is_all_spaces = s.chars().all(|c| c == ' ');

    let mut result = String::with_capacity(s.len() * 3);
    for c in s.chars() {
        if c == ' ' {
            if is_all_spaces {
                // Only use space marker if entire string is spaces
                result.push_str("《_》");
            } else {
                result.push(c);
            }
        } else if let Some((_, replacement)) = WHITESPACE_MARKERS.iter().find(|(ws, _)| *ws == c) {
            result.push_str(replacement);
        } else {
            result.push(c);
        }
    }
    result
}

/// Converts a byte slice to a `Cow<str>` for output formatting.
///
/// Uses SIMD-accelerated UTF-8 validation via `simdutf8::basic::from_utf8` on the
/// happy path (returns a borrowed `&str` with no allocation on valid UTF-8, which
/// is the overwhelming common case including all-ASCII).
///
/// Falls back to `String::from_utf8_lossy` only when the bytes are not valid
/// UTF-8 (replacement characters substituted, allocates).
///
/// This is faster than `String::from_utf8_lossy` alone, which always uses scalar
/// UTF-8 validation and pre-allocates a `String` of the input length.
#[inline]
pub fn bytes_to_cow_str(c: &[u8]) -> std::borrow::Cow<'_, str> {
    if let Ok(s) = simdutf8::basic::from_utf8(c) {
        std::borrow::Cow::Borrowed(s)
    } else {
        String::from_utf8_lossy(c)
    }
}

pub fn qsv_custom_panic() {
    setup_panic!(
        human_panic::Metadata::new(env!("CARGO_PKG_NAME"), env!("CARGO_PKG_VERSION"))
            .authors("datHere qsv maintainers")
            .homepage("https://qsv.dathere.com")
            .support("- Open a GitHub issue at https://github.com/dathere/qsv/issues")
    );
}

fn default_user_agent() -> String {
    let unknown_command = "Unknown".to_string();
    let current_command = CURRENT_COMMAND.get().unwrap_or(&unknown_command);
    format!(
        "{CARGO_BIN_NAME}/{CARGO_PKG_VERSION} ({TARGET}; {current_command}; {QSV_KIND}; https://github.com/dathere/qsv)"
    )
}

pub fn max_jobs() -> usize {
    let num_cpus = num_cpus();
    let max_jobs = match env::var("QSV_MAX_JOBS") {
        Ok(val) => val.parse::<usize>().unwrap_or(1_usize),
        Err(_) => num_cpus,
    };
    if (1..=num_cpus).contains(&max_jobs) {
        max_jobs
    } else {
        num_cpus
    }
}

/// Given a desired number of cores to use
/// returns number of cores to actually use and set
/// rayon global thread pool size accordingly.
/// If desired is None, zero, or greater than available cores,
/// returns `max_jobs`, which is equal to number of available cores
/// If desired is Some and less than available cores,
/// returns desired number of cores
pub fn njobs(flag_jobs: Option<usize>) -> usize {
    let njobs_result = JOBS_TO_USE.get_or_init(|| {
        let max_jobs = max_jobs();
        let jobs_to_use = flag_jobs.map_or(max_jobs, |jobs| {
            if jobs == 0 || jobs > max_jobs {
                max_jobs
            } else {
                jobs
            }
        });
        match rayon::ThreadPoolBuilder::new()
            .num_threads(jobs_to_use)
            .build_global()
        {
            Err(e) => {
                log::warn!("Failed to set global thread pool size to {jobs_to_use}: {e}");
            },
            _ => {
                log::info!("Using {jobs_to_use} jobs...");
            },
        }
        jobs_to_use
    });
    *njobs_result
}

/// One-time, process-wide allocator setup. Called from `main()` right after the
/// signal handlers are installed, before any command runs.
///
/// Lever A: enable jemalloc `background_thread`, which offloads page purging
/// (`madvise`) to background threads, removing it from the foreground hot path at
/// NO RSS cost. Supported on Linux; rejected on macOS (the write errors — handled
/// best-effort, leaving `BACKGROUND_THREADS_ACTIVE` false). The per-command
/// retention lever (Lever B) is itself gated to Linux, so it does NOT engage on
/// macOS as a fallback. Disabled entirely by `QSV_NO_ALLOC_TUNING`. No
/// `cfg(target_os)` gate: the write+read-back is self-correcting across platforms.
#[cfg(all(feature = "jemallocator", not(feature = "mimalloc")))]
pub fn init_allocator_runtime() {
    use std::sync::atomic::Ordering;

    use tikv_jemalloc_ctl::background_thread;

    if get_envvar_flag("QSV_NO_ALLOC_TUNING") {
        log::info!("alloc tuning disabled via QSV_NO_ALLOC_TUNING");
        return;
    }
    // Best-effort: ignore the write error (e.g. macOS, where it's unsupported)
    // and trust the read-back for the authoritative state.
    let _ = background_thread::write(true);
    if let Ok(active) = background_thread::read() {
        BACKGROUND_THREADS_ACTIVE.store(active, Ordering::Relaxed);
        log::info!("jemalloc background_thread active: {active}");
    }
}

#[cfg(not(all(feature = "jemallocator", not(feature = "mimalloc"))))]
pub fn init_allocator_runtime() {}

/// PROTOTYPE (jemalloc per-command tuning): aggregation-heavy commands
/// (`stats -E`, `frequency`, ...) build many `Frequencies` hashmaps that grow and
/// merge across rayon workers, producing heavy `madvise` page-return churn under
/// jemalloc's default decay. Retaining freed pages (dirty/muzzy decay = -1) for
/// the duration keeps them mapped for reuse, trading higher peak RSS (~+16% on a
/// 514 MB input) for fewer syscalls (~6-9% faster `stats -E`). Scoped per-command
/// — qsv is short-lived, so the process exits before retained pages matter; no
/// restore is performed.
///
/// LINUX-ONLY: this is only a win where jemalloc's default `madvise`-based page
/// decay creates the foreground purge churn it removes. On macOS, Lever A
/// (`background_thread`) is unsupported, so this fallback would otherwise run —
/// but the macOS memory compressor turns page retention into a net loss
/// (measured on a 32 GB host: `moarstats --bivariate-stats all` peak footprint
/// +23% and peak RSS ~32 GB, tipping into swap for a 40-60% slowdown; even on a
/// 64 GB host where it fits, ~+10 GB footprint, ~1.7x system time, and ~6%
/// slower wall-clock, with byte-identical output). So it is gated off there.
///
/// NOTE: the `MALLCTL_ARENAS_ALL` (4096) pseudo-index segfaults here for decay
/// writes via BOTH the string-name (`arena.4096.*`) and numeric-MIB interfaces
/// (tikv-jemalloc-sys 0.7 / macOS), so all arenas can't be set in one call.
/// Instead set the `arenas.*` template (inherited by arenas created later, e.g.
/// lazily-bound rayon workers) and loop over the currently-initialized arenas by
/// index. This is timing-dependent — arenas not yet created when this runs are
/// covered only by the template — but benchmarks show it captures essentially all
/// of what `opt.dirty_decay_ms` (the env-var path) delivers, within noise.
#[cfg(all(feature = "jemallocator", not(feature = "mimalloc")))]
pub fn retain_alloc_pages_for_aggregation() {
    use std::sync::atomic::Ordering;

    use tikv_jemalloc_ctl::raw;

    // Skip when tuning is disabled, or when Lever A (background_thread) is active —
    // background purging already removes the madvise churn without the RSS cost of
    // retaining pages, so applying retention on top would be pure RSS loss.
    if get_envvar_flag("QSV_NO_ALLOC_TUNING") || BACKGROUND_THREADS_ACTIVE.load(Ordering::Relaxed) {
        return;
    }
    // Retention only pays off on Linux (see fn doc). On macOS the memory
    // compressor makes it a net loss, and macOS can't run Lever A to offset it,
    // so the fallback would otherwise misfire here.
    if !cfg!(target_os = "linux") {
        return;
    }
    // safety: tikv_jemalloc_ctl::raw {read,write} call jemalloc's mallctl by name.
    // Invariants upheld here:
    //   - every key is a NUL-terminated byte string (the trailing `\0`), as the name-based mallctl
    //     API requires;
    //   - value widths match the mallctl types: `*_decay_ms` is `ssize_t` (isize, -1 = "never
    //     decay/retain pages"); `arenas.narenas` is `unsigned` (u32);
    //   - every result is discarded best-effort (`let _ =`), so a jemalloc build or platform
    //     lacking a given knob returns an error rather than aborting the command. No pointers are
    //     passed; mallctl copies the value in/out.
    unsafe {
        let _ = raw::write(b"arenas.dirty_decay_ms\0", -1_isize);
        let _ = raw::write(b"arenas.muzzy_decay_ms\0", -1_isize);
        if let Ok(narenas) = raw::read::<u32>(b"arenas.narenas\0") {
            for i in 0..narenas {
                let _ = raw::write(format!("arena.{i}.dirty_decay_ms\0").as_bytes(), -1_isize);
                let _ = raw::write(format!("arena.{i}.muzzy_decay_ms\0").as_bytes(), -1_isize);
            }
        }
    }
}

#[cfg(not(all(feature = "jemallocator", not(feature = "mimalloc"))))]
pub fn retain_alloc_pages_for_aggregation() {}

pub fn timeout_secs(timeout: u16) -> Result<u64, String> {
    let timeout = match env::var("QSV_TIMEOUT") {
        Ok(val) => val.parse::<u16>().unwrap_or(30_u16),
        Err(_) => timeout,
    };

    if timeout > 3600 {
        return fail_format!("Timeout cannot be more than 3,600 seconds (1 hour): {timeout}");
    } else if timeout == 0 {
        return fail!("Timeout cannot be zero.");
    }
    log::info!("TIMEOUT: {timeout}");
    Ok(timeout as u64)
}

/// sets custom user agent
/// if user agent is not set, then use the default user agent
/// it supports four special LITERALs: $`QSV_BIN_NAME`, $`QSV_VERSION`, $`QSV_TARGET`, $`QSV_KIND`
/// and $`QSV_COMMAND` which will be replaced with the actual values during runtime
pub fn set_user_agent(user_agent: Option<String>) -> CliResult<String> {
    use reqwest::header::HeaderValue;

    let ua = match user_agent {
        Some(ua_arg) => ua_arg,
        None => env::var("QSV_USER_AGENT").unwrap_or_else(|_| default_user_agent()),
    };

    let unknown_command = "Unknown".to_string();
    let current_command = CURRENT_COMMAND.get().unwrap_or(&unknown_command);

    // look for special literals - $QSV_VERSION and $QSV_TARGET and replace them
    let ua = ua
        .replace("$QSV_BIN_NAME", CARGO_BIN_NAME)
        .replace("$QSV_VERSION", CARGO_PKG_VERSION)
        .replace("$QSV_TARGET", TARGET)
        .replace("$QSV_KIND", QSV_KIND)
        .replace("$QSV_COMMAND", current_command);

    match HeaderValue::from_str(ua.as_str()) {
        Ok(_) => (),
        Err(e) => return fail_incorrectusage_clierror!("Invalid user-agent value: {e}"),
    }

    log::info!("set user agent: {ua}");
    Ok(ua)
}

/// Creates a standardized async reqwest client with common configuration options.
/// This function centralizes the client creation logic used across multiple commands.
///
/// # Arguments
///
/// * `user_agent` - Optional custom user agent string
/// * `timeout_secs` - Inactivity (idle) timeout in seconds for HTTP requests. If 0, no timeout.
/// * `base_url` - Optional base URL for retry configuration
///
/// # Returns
///
/// Returns a configured reqwest client with:
/// - Custom user agent (or default)
/// - Compression support (brotli, gzip, deflate, zstd)
/// - Rustls TLS backend
/// - HTTP/2 adaptive window
/// - Connection verbose logging (when debug/trace enabled)
/// - Idle/read timeout configuration (NOT a total-request deadline)
/// - Retry logic for service unavailable errors
pub fn create_reqwest_async_client(
    user_agent: Option<String>,
    timeout_secs: u16,
    base_url: Option<String>,
) -> CliResult<Client> {
    let base_url_for_retry = base_url.unwrap_or_default();

    let retries = reqwest::retry::for_host(base_url_for_retry).classify_fn(|req_rep| {
        if req_rep.status() == Some(reqwest::StatusCode::SERVICE_UNAVAILABLE) {
            req_rep.retryable()
        } else {
            req_rep.success()
        }
    });

    let mut builder = Client::builder()
        .user_agent(set_user_agent(user_agent)?)
        .brotli(true)
        .gzip(true)
        .deflate(true)
        .zstd(true)
        .use_rustls_tls()
        .http2_adaptive_window(true)
        .connection_verbose(log_enabled!(log::Level::Debug) || log_enabled!(log::Level::Trace))
        .retry(retries);

    if timeout_secs > 0 {
        // Use an INACTIVITY (idle) timeout, not a total-request deadline. `read_timeout`
        // applies to each read of the response body and RESETS after every successful read,
        // so a slow-but-steady download (e.g. a large CSV trickling in over a slow proxy) is
        // never aborted mid-stream — only a genuinely stalled connection (no bytes for
        // `timeout_secs`) trips it. `connect_timeout` still bounds a stalled connect/TLS
        // handshake (not reliably caught otherwise on flaky windows-arm64 CI runners).
        // There is deliberately no total-request `.timeout()` cap.
        let timeout_duration = Duration::from_secs(timeout_secs.into());
        builder = builder
            .read_timeout(timeout_duration)
            .connect_timeout(timeout_duration);
    }

    Ok(builder.build()?)
}

/// Creates a standardized blocking reqwest client with common configuration options.
/// This function centralizes the blocking client creation logic used across multiple commands.
///
/// # Arguments
///
/// * `user_agent` - Optional custom user agent string
/// * `timeout_secs` - Timeout in seconds for HTTP requests. If 0, no timeout is used.
/// * `base_url` - Optional base URL for retry configuration
///
/// # Returns
///
/// Returns a configured blocking reqwest client with the same options
/// as `create_reqwest_async_client`
///
/// NOTE: unlike the async client, this uses a TOTAL-request `.timeout()`, not an inactivity
/// (`read_timeout`) one. reqwest 0.13.4's blocking client reads the response body on the
/// calling thread via a park/unpark waker with NO Tokio timer running, so a `read_timeout`
/// body wrapper panics ("there is no reactor running") the moment the body is read. The
/// idle-timeout semantics therefore apply only to `create_reqwest_async_client`.
pub fn create_reqwest_blocking_client(
    user_agent: Option<String>,
    timeout_secs: u16,
    base_url: Option<String>,
) -> CliResult<reqwest::blocking::Client> {
    let timeout_duration = Duration::from_secs(timeout_secs.into());
    let base_url_for_retry = base_url.unwrap_or_default();

    let retries = reqwest::retry::for_host(base_url_for_retry).classify_fn(|req_rep| {
        if req_rep.status() == Some(reqwest::StatusCode::SERVICE_UNAVAILABLE) {
            req_rep.retryable()
        } else {
            req_rep.success()
        }
    });

    let mut builder = reqwest::blocking::Client::builder()
        .user_agent(set_user_agent(user_agent)?)
        .brotli(true)
        .gzip(true)
        .deflate(true)
        .zstd(true)
        .use_rustls_tls()
        .http2_adaptive_window(true)
        .connection_verbose(log_enabled!(log::Level::Debug) || log_enabled!(log::Level::Trace))
        .retry(retries);

    if timeout_secs > 0 {
        // bound the total request AND the connect/TLS handshake separately.
        // a stalled connect is not reliably caught by the total timeout alone
        // (e.g. flaky windows-arm64 CI runners), so set connect_timeout too.
        builder = builder
            .timeout(timeout_duration)
            .connect_timeout(timeout_duration);
    } else {
        // `timeout_secs == 0` means "no timeout" to every caller of this function (see
        // callers computing 0 to disable timeouts for local LLM endpoints), but
        // reqwest::blocking::ClientBuilder does NOT default to "no timeout" when
        // `.timeout()` is simply never called: its internal `Timeout` wrapper has its own
        // `Default` impl of `Some(Duration::from_secs(30))` (reqwest 0.13.4
        // src/blocking/client.rs), independent of the inner async client's `Config` (which
        // DOES default to `None`/no timeout). Skipping `.timeout()` here silently left
        // that 30s default active, killing legitimate long-running local-LLM completions
        // (e.g. describegpt --dictionary against LM Studio) exactly 30s in. Explicitly
        // passing `None` overrides it.
        builder = builder.timeout(None);
    }

    Ok(builder.build()?)
}

/// Transforms a GitHub blob URL to a raw.githubusercontent.com URL.
/// Returns the original URL unchanged if not a GitHub blob URL.
///
/// Query parameters (e.g., `?ref=v1.0`) and fragments (e.g., `#L10-L20`) are stripped
/// from the transformed URL since they don't apply to raw.githubusercontent.com.
///
/// Note: This only works for github.com URLs. GitHub Enterprise URLs
/// (e.g., `https://github.company.com/...`) are not transformed.
///
/// # Arguments
///
/// * `url` - The URL to potentially transform
///
/// # Returns
///
/// A String containing the transformed URL (for GitHub blob URLs)
/// or the original URL (for all other URLs).
///
/// # Examples
///
/// ```
/// // GitHub blob URLs are transformed to raw URLs:
/// // https://github.com/user/repo/blob/branch/path/file.csv
/// // becomes:
/// // https://raw.githubusercontent.com/user/repo/branch/path/file.csv
///
/// // Non-GitHub URLs are returned unchanged
/// ```
#[inline]
pub fn transform_github_url(url: &str) -> String {
    // Pattern: https://github.com/USER/REPO/blob/BRANCH/PATH
    // Transform: https://raw.githubusercontent.com/USER/REPO/BRANCH/PATH

    let github_prefix_https = "https://github.com/";
    let github_prefix_http = "http://github.com/";

    let remainder = if let Some(r) = url.strip_prefix(github_prefix_https) {
        r
    } else if let Some(r) = url.strip_prefix(github_prefix_http) {
        r
    } else {
        return url.to_string();
    };

    if let Some(blob_idx) = remainder.find("/blob/") {
        let user_repo = &remainder[..blob_idx];
        let mut branch_and_path = &remainder[blob_idx + 6..]; // skip "/blob/"

        // Strip query parameters and fragments - they don't apply to raw.githubusercontent.com
        if let Some(query_idx) = branch_and_path.find('?') {
            branch_and_path = &branch_and_path[..query_idx];
        }
        if let Some(fragment_idx) = branch_and_path.find('#') {
            branch_and_path = &branch_and_path[..fragment_idx];
        }

        let raw_url = format!("https://raw.githubusercontent.com/{user_repo}/{branch_and_path}");
        log::info!("Transformed GitHub blob URL to raw URL: {raw_url}");
        raw_url
    } else {
        url.to_string()
    }
}

pub fn version() -> String {
    let mut enabled_features = String::new();

    #[cfg(all(feature = "apply", not(feature = "lite")))]
    enabled_features.push_str("apply;");
    #[cfg(all(feature = "fetch", not(feature = "lite")))]
    enabled_features.push_str("fetch;");
    #[cfg(all(feature = "foreach", not(feature = "lite")))]
    enabled_features.push_str("foreach;");
    #[cfg(all(feature = "geocode", not(feature = "lite")))]
    enabled_features.push_str("geocode;");

    #[cfg(all(feature = "luau", not(feature = "lite")))]
    {
        let luau = mlua::Lua::new();
        match luau.load("return _VERSION").eval() {
            Ok(version_info) => {
                match version_info {
                    mlua::Value::String(luaustring_val) => {
                        let string_val = luaustring_val.to_string_lossy();
                        if string_val == "Luau" {
                            enabled_features.push_str("Luau - version not specified;");
                        } else {
                            // safety: safe to unwrap as we're just using it to append to
                            // enabled_features
                            write!(enabled_features, "{string_val};").unwrap();
                        }
                    },
                    _ => {
                        enabled_features.push_str("Luau - ?;");
                    },
                }
            },
            // safety: safe to unwrap as we're just using it to append to enabled_features
            Err(e) => write!(enabled_features, "Luau - cannot retrieve version: {e};").unwrap(),
        }
    }
    #[cfg(all(feature = "magika", feature = "feature_capable"))]
    enabled_features.push_str("magika;");
    #[cfg(all(feature = "prompt", feature = "feature_capable"))]
    enabled_features.push_str("prompt;");

    #[cfg(all(feature = "python", not(feature = "lite")))]
    {
        enabled_features.push_str("python-");
        pyo3::Python::attach(|_| {
            enabled_features.push_str(pyo3::Python::version_str());
            enabled_features.push(';');
        });
    }
    #[cfg(all(feature = "to", not(feature = "lite")))]
    enabled_features.push_str("to;");
    #[cfg(all(feature = "viz", not(feature = "lite")))]
    enabled_features.push_str("viz;");
    #[cfg(all(feature = "viz_static", not(feature = "lite")))]
    enabled_features.push_str("viz_static;");
    #[allow(clippy::const_is_empty)]
    #[cfg(all(feature = "polars", not(feature = "lite")))]
    if QSV_POLARS_REV.is_empty() {
        enabled_features.push_str(format!("polars-{};", polars::VERSION).as_str());
    } else {
        enabled_features
            .push_str(format!("polars-{}:{};", polars::VERSION, QSV_POLARS_REV).as_str());
    }
    #[cfg(feature = "self_update")]
    enabled_features.push_str("self_update");
    enabled_features.push('-');

    // get max_file_size & memory info. max_file_size is based on QSV_FREEMEMORY_HEADROOM_PCT
    // setting and is only enforced when qsv is running in "non-streaming" mode (i.e. needs to
    // load the entire file into memory).
    let mut sys = System::new();
    sys.refresh_memory();
    let avail_mem = sys.available_memory();
    let total_mem = sys.total_memory();
    let free_swap = sys.free_swap();
    let max_file_size = mem_file_check(Path::new(""), true, false).unwrap_or(0) as u64;

    // we also get System info to help with debugging and logging.
    // we just need the CPU model name, so we ask for nothing
    sys.refresh_cpu_list(sysinfo::CpuRefreshKind::nothing());
    let os_version = System::long_os_version().unwrap_or_else(|| "Unknown".to_string());
    let kernel_version = System::kernel_long_version();
    let cpu_brand = sys
        .cpus()
        .first()
        .map_or("Unknown", |cpu| cpu.brand().trim());
    let physical_cpu_count = System::physical_core_count().unwrap_or(0);

    #[cfg(all(feature = "mimalloc", not(feature = "jemallocator")))]
    let malloc_kind = {
        let mimalloc_version = mimalloc::MiMalloc.version();
        format!("mimalloc {mimalloc_version}")
    };
    #[cfg(all(feature = "jemallocator", not(feature = "mimalloc")))]
    let malloc_kind = if BACKGROUND_THREADS_ACTIVE.load(std::sync::atomic::Ordering::Relaxed) {
        "jemalloc+bgthread"
    } else {
        "jemalloc"
    };
    #[cfg(not(any(feature = "mimalloc", feature = "jemallocator")))]
    let malloc_kind = "standard";
    let (qsvtype, maj, min, pat, pre, rustversion) = (
        option_env!("CARGO_BIN_NAME"),
        option_env!("CARGO_PKG_VERSION_MAJOR"),
        option_env!("CARGO_PKG_VERSION_MINOR"),
        option_env!("CARGO_PKG_VERSION_PATCH"),
        option_env!("CARGO_PKG_VERSION_PRE"),
        option_env!("CARGO_PKG_RUST_VERSION"),
    );
    if let (Some(qsvtype), Some(maj), Some(min), Some(pat), Some(pre), Some(rustversion)) =
        (qsvtype, maj, min, pat, pre, rustversion)
    {
        if pre.is_empty() {
            format!(
                "{qsvtype} {maj}.{min}.{pat}-{malloc_kind}-{enabled_features}{maxjobs}-{numcpus};\
                 {max_file_size}-{free_swap}-{avail_mem}-{total_mem} ({TARGET} compiled with Rust \
                 {rustversion};{os_version}-{kernel_version};{cpu_brand}-{physical_cpu_count}) \
                 {QSV_KIND}",
                maxjobs = max_jobs(),
                numcpus = num_cpus(),
                max_file_size = indicatif::HumanBytes(max_file_size),
                free_swap = indicatif::HumanBytes(free_swap),
                avail_mem = indicatif::HumanBytes(avail_mem),
                total_mem = indicatif::HumanBytes(total_mem),
            )
        } else {
            format!(
                "{qsvtype} {maj}.{min}.\
                 {pat}-{pre}-{malloc_kind}-{enabled_features}{maxjobs}-{numcpus};\
                 {max_file_size}-{free_swap}-{avail_mem}-{total_mem} ({TARGET} compiled with Rust \
                 {rustversion}) {QSV_KIND}",
                maxjobs = max_jobs(),
                numcpus = num_cpus(),
                max_file_size = indicatif::HumanBytes(max_file_size),
                free_swap = indicatif::HumanBytes(free_swap),
                avail_mem = indicatif::HumanBytes(avail_mem),
                total_mem = indicatif::HumanBytes(total_mem),
            )
        }
    } else {
        String::new()
    }
}

const OTHER_ENV_VARS: &[&str] = &["all_proxy", "no_proxy", "http_proxy", "https_proxy"];

pub fn show_env_vars() {
    let mut env_var_set = false;
    for (n, v) in env::vars_os() {
        // safety: we know that the env::vars_os() will not fail
        let env_var = n.into_string().unwrap();
        #[cfg(all(feature = "mimalloc", not(feature = "jemallocator")))]
        if env_var.starts_with("QSV_")
            || env_var.starts_with("MIMALLOC_")
            || OTHER_ENV_VARS.contains(&env_var.to_ascii_lowercase().as_str())
        {
            env_var_set = true;
            woutinfo!("{env_var}: {v:?}");
        }
        #[cfg(all(feature = "jemallocator", not(feature = "mimalloc")))]
        if env_var.starts_with("QSV_")
            || env_var.starts_with("JEMALLOC_")
            || env_var == "MALLOC_CONF"
            || env_var == "_RJEM_MALLOC_CONF"
            || OTHER_ENV_VARS.contains(&env_var.to_ascii_lowercase().as_str())
        {
            env_var_set = true;
            woutinfo!("{env_var}: {v:?}");
        }
        #[cfg(not(any(feature = "mimalloc", feature = "jemallocator")))]
        if env_var.starts_with("QSV_")
            || OTHER_ENV_VARS.contains(&env_var.to_ascii_lowercase().as_str())
        {
            env_var_set = true;
            woutinfo!("{env_var}: {v:?}");
        }
        #[cfg(feature = "polars")]
        if env_var.starts_with("POLARS_") {
            env_var_set = true;
            woutinfo!("{env_var}: {v:?}");
        }
    }
    if !env_var_set {
        woutinfo!("No qsv-relevant environment variables set.");
    }
}

#[inline]
pub fn count_rows(conf: &Config) -> Result<u64, CliError> {
    // Check if ROW_COUNT is already initialized to avoid redundant counting
    if let Some(count) = ROW_COUNT.get() {
        return Ok(count.unwrap_or(0));
    }

    // If not, try using index if available
    if let Some(idx) = conf.indexed().unwrap_or(None) {
        return Ok(idx.count());
    }
    // index does not exist or is stale

    // Otherwise, count records by using polars mem-mapped reader if available
    // If polars is not enabled, count records by iterating through records
    // Do this only once per invocation and cache the result in ROW_COUNT,
    // so we don't have to re-count rows every time we need to know the
    // rowcount for CSVs that don't have an index.
    ROW_COUNT
        .get_or_init(|| {
            // Try different counting methods in order of preference
            count_rows_with_best_method(conf)
        })
        .ok_or_else(|| CliError::Other("Unable to get row count".to_string()))
}

#[cfg(feature = "polars")]
fn count_rows_with_best_method(conf: &Config) -> Option<u64> {
    if !conf.no_headers {
        // Try polars first for files with headers
        if let Ok(polars_count) = polars_count_input(conf, false) {
            // If count is greater than 0, return the polars accelerated count
            // as sometimes, polars returns a zero count even if the file is not empty
            // and the file is a proper CSV file.
            // Otherwise, double-check with the "regular" CSV reader
            if polars_count > 0 {
                return Some(polars_count);
            }
        }
    }

    // Fall back to CSV reader
    count_with_csv_reader(conf)
}

#[cfg(not(feature = "polars"))]
fn count_rows_with_best_method(conf: &Config) -> Option<u64> {
    count_with_csv_reader(conf)
}

fn count_with_csv_reader(conf: &Config) -> Option<u64> {
    let path_display = conf
        .path
        .as_deref()
        .map_or_else(|| "<stdin>".to_string(), |p| p.display().to_string());
    conf.clone()
        .skip_format_check(true)
        .reader()
        .ok()
        .map(|mut rdr| {
            let mut count = 0_u64;
            let mut record = csv::ByteRecord::new();
            loop {
                match rdr.read_byte_record(&mut record) {
                    Ok(true) => count += 1,
                    Ok(false) => break,
                    Err(e) => {
                        log::warn!(
                            "CSV read error during row count of {path_display} after {count} \
                             rows; returning partial count: {e}"
                        );
                        break;
                    },
                }
            }
            count
        })
}

/// Count rows using the "regular" CSV reader, never the polars mem-mapped
/// reader. The result is cached in `ROW_COUNT` so subsequent calls are free.
#[inline]
pub fn count_rows_regular(conf: &Config) -> Result<u64, CliError> {
    if let Some(idx) = conf.indexed().unwrap_or(None) {
        return Ok(idx.count());
    }
    // No index (or stale): defer to the shared CSV-reader helper so the
    // counting loop and warn-with-path logic live in one place. Cached in
    // ROW_COUNT so a later count_rows call hits the same value.
    ROW_COUNT
        .get_or_init(|| count_with_csv_reader(conf))
        .ok_or_else(|| CliError::Other("Unable to get row count".to_string()))
}

#[cfg(any(feature = "feature_capable", feature = "lite"))]
pub fn count_lines_in_file(file: &str) -> Result<u64, CliError> {
    let file = File::open(file)?;
    let reader = BufReader::new(file);

    let line_count = reader.lines().count() as u64;
    Ok(line_count)
}

pub fn prep_progress(progress: &ProgressBar, record_count: u64) {
    progress.set_style(
        ProgressStyle::default_bar()
            .template("[{elapsed_precise}] [{wide_bar} {percent}%{msg}] ({per_sec} - {eta})")
            .unwrap(),
    );
    progress.set_message(format!(" of {} records", HumanCount(record_count)));

    // draw progress bar for the first time using specified style
    progress.set_length(record_count);

    log::info!("Progress started... {record_count} records");
}

pub fn finish_progress(progress: &ProgressBar) {
    progress.set_style(
        ProgressStyle::default_bar()
            .template("[{elapsed_precise}] [{wide_bar} {percent}%{msg}] ({per_sec})")
            .unwrap(),
    );

    if progress.length().unwrap_or_default() == progress.position() {
        progress.finish();
        log::info!("Progress done... {}", progress.message());
    } else {
        progress.abandon();
        log::info!("Progress abandoned... {}", progress.message());
    }
}

#[cfg(all(feature = "fetch", not(feature = "lite")))]
macro_rules! update_cache_info {
    ($progress:expr_2021, $cache_instance:expr_2021) => {
        use cached::Cached;
        use indicatif::HumanCount;

        let cache = $cache_instance.read();
        let size = cache.cache_size();
        if size > 0 {
            let hits = cache.cache_hits().unwrap_or_default();
            let misses = cache.cache_misses().unwrap_or(1);
            #[allow(clippy::cast_precision_loss)]
            let hit_ratio = (hits as f64 / (hits + misses) as f64) * 100.0;
            let capacity = cache.cache_capacity();
            drop(cache);
            $progress.set_message(format!(
                " of {} records. Cache {:.2}% entries: {} capacity: {}.",
                HumanCount($progress.length().unwrap()),
                hit_ratio,
                HumanCount(size as u64),
                HumanCount(capacity.unwrap() as u64),
            ));
        }
    };
    ($progress:expr_2021, $cache_hits:expr_2021, $num_rows:expr_2021) => {
        use indicatif::HumanCount;

        #[allow(clippy::cast_precision_loss)]
        let hit_ratio = ($cache_hits as f64 / $num_rows as f64) * 100.0;
        $progress.set_message(format!(
            " of {} records. Cache hit ratio: {hit_ratio:.2}%",
            HumanCount($progress.length().unwrap()),
        ));
    };
}

#[cfg(all(feature = "fetch", not(feature = "lite")))]
pub(crate) use update_cache_info;

pub fn get_args<T>(usage: &str, argv: &[&str]) -> CliResult<T>
where
    T: DeserializeOwned,
{
    Docopt::new(usage)
        .and_then(|d| {
            d.argv(argv.iter().copied())
                .version(Some(version()))
                .deserialize()
        })
        .map_err(From::from)
}

#[inline]
pub fn many_configs(
    inps: &[PathBuf],
    delim: Option<Delimiter>,
    no_headers: bool,
    flexible: bool,
) -> Result<Vec<Config>, String> {
    let mut inps = inps
        .iter()
        .map(|p| p.to_str().unwrap_or("-").to_owned())
        .collect::<Vec<_>>();
    if inps.is_empty() {
        inps.push("-".to_owned()); // stdin
    }
    let confs = inps
        .into_iter()
        .map(|p| {
            Config::new(Some(p).as_ref())
                .delimiter(delim)
                .no_headers(no_headers)
                .flexible(flexible)
        })
        .collect::<Vec<_>>();
    errif_greater_one_stdin(&confs)?;
    Ok(confs)
}

pub fn errif_greater_one_stdin(inps: &[Config]) -> Result<(), String> {
    let nstd = inps.iter().filter(|inp| inp.is_stdin()).count();
    if nstd > 1 {
        return fail!("At most one <stdin> input is allowed.");
    }
    Ok(())
}

pub const fn chunk_size(nitems: usize, njobs: usize) -> usize {
    if nitems < njobs {
        nitems
    } else {
        nitems / njobs
    }
}

pub const fn num_of_chunks(nitems: usize, chunk_size: usize) -> usize {
    if chunk_size == 0 {
        return nitems;
    }
    let mut n = nitems / chunk_size;
    if !nitems.is_multiple_of(chunk_size) {
        n += 1;
    }
    n
}

pub fn file_metadata(md: &fs::Metadata) -> (u64, u64) {
    use filetime::FileTime;
    let last_modified = FileTime::from_last_modification_time(md).unix_seconds() as u64;
    let fsize = md.len();
    (last_modified, fsize)
}

/// Check if there is enough memory to process the file.
/// Return the maximum file size that can be processed.
/// If the file is larger than the maximum file size, return an error.
/// If memcheck is true, check memory in CONSERVATIVE mode
///   (i.e., Filesize < (AVAIL memory + SWAP) * `platform_factor` - headroom)
/// If memcheck is false, check memory in NORMAL mode
///   (i.e., Filesize < TOTAL memory - headroom)
pub fn mem_file_check(
    path: &Path,
    version_check: bool,
    conservative_memcheck: bool,
) -> CliResult<i64> {
    // if we're NOT calling this from the version() and the file doesn't exist,
    // we don't need to check memory as file existence is checked before this function is called.
    // If we do get here with a non-existent file, that means we're using stdin,
    // so this check doesn't apply, so we return -1
    if !path.exists() && !version_check {
        return Ok(-1_i64);
    }

    let conservative_memcheck_work = get_envvar_flag("QSV_MEMORY_CHECK") || conservative_memcheck;

    let mut mem_pct =
        env::var("QSV_FREEMEMORY_HEADROOM_PCT").map_or(DEFAULT_FREEMEMORY_HEADROOM_PCT, |val| {
            atoi_simd::parse::<u8, false, false>(val.as_bytes())
                .unwrap_or(DEFAULT_FREEMEMORY_HEADROOM_PCT)
        });

    // if QSV_FREEMEMORY_HEADROOM_PCT is 0, we skip the memory check
    if mem_pct == 0 {
        return Ok(i64::MAX);
    }

    let mut sys = sysinfo::System::new();
    sys.refresh_memory();
    let avail_mem = sys.available_memory();
    let free_swap = sys.free_swap();
    let total_mem = sys.total_memory();

    // for safety, we don't want to go below 10% memory headroom
    // nor above 90% memory headroom as its too memory-restrictive
    mem_pct = mem_pct.clamp(10, 90);

    // Platform-specific adjustment factors for conservative mode
    // These account for OS-specific memory management capabilities
    let platform_factor = cfg_select! {
        // macOS has aggressive memory compression & dynamic swap
        target_os = "macos" => 1.3,
        // Linux page cache is reclaimable, but be conservative
        target_os = "linux" => 1.15,
        // Windows memory reporting is already conservative
        target_os = "windows" => 1.0,
        // Other platforms: no adjustment
        _ => 1.0,
    };

    // Calculate maximum available memory based on mode
    #[allow(clippy::cast_precision_loss)]
    let max_avail_mem = if conservative_memcheck_work {
        // CONSERVATIVE: Use available + swap, with platform adjustments
        let base_mem = (avail_mem + free_swap) as f64;
        let adjusted_mem = base_mem * platform_factor;
        (adjusted_mem * ((100 - mem_pct) as f64 / 100.0)) as u64
    } else {
        // NORMAL: Use total memory
        (total_mem as f64 * ((100 - mem_pct) as f64 / 100.0)) as u64
    };

    // if we're calling this from version(), we don't need to check the file size
    if !version_check {
        let file_metadata =
            fs::metadata(path).map_err(|e| format!("Failed to get file size: {e}"))?;
        let fsize = file_metadata.len();
        if fsize > max_avail_mem {
            return fail_oom_clierror!(
                "Not enough memory to process the file. qsv running in non-streaming {mode} mode. \
                 Total memory: {total_mem} Available memory: {avail_mem}. Free swap: {free_swap} \
                 Max Available memory/Max input file size: {max_avail_mem}. \
                 QSV_FREEMEMORY_HEADROOM_PCT: {mem_pct}%. File size: {fsize}.",
                mode = if conservative_memcheck_work {
                    "CONSERVATIVE"
                } else {
                    "NORMAL"
                },
                total_mem = indicatif::HumanBytes(total_mem),
                avail_mem = indicatif::HumanBytes(avail_mem),
                free_swap = indicatif::HumanBytes(free_swap),
                max_avail_mem = indicatif::HumanBytes(max_avail_mem),
                mem_pct = mem_pct,
                fsize = indicatif::HumanBytes(fsize)
            );
        }
    }

    Ok(max_avail_mem as i64)
}

#[cfg(any(feature = "feature_capable", feature = "lite"))]
#[inline]
pub fn condense(val: Cow<[u8]>, n: Option<usize>) -> Cow<[u8]> {
    let Some(n) = n else { return val };

    if let Ok(s) = simdutf8::basic::from_utf8(&val) {
        // Valid UTF-8: count by chars, not bytes, so multi-byte codepoints
        // count once. Truncating by bytes here would risk splitting a char.
        if s.chars().count() <= n {
            return val;
        }
        let mut out: String = s.chars().take(n).collect();
        out.push_str("...");
        return Cow::Owned(out.into_bytes());
    }

    // Not valid UTF-8: truncate by bytes.
    if val.len() <= n {
        return val;
    }
    let mut out = val[..n].to_vec();
    out.extend_from_slice(b"...");
    Cow::Owned(out)
}

pub fn idx_path(csv_path: &Path) -> PathBuf {
    // safety: we know the path has a filename
    let mut p = csv_path
        .to_path_buf()
        .into_os_string()
        .into_string()
        .unwrap();
    p.push_str(".idx");
    PathBuf::from(&p)
}

/// Creates an index file for the given CSV file path.
///
/// This function creates a CSV index file that enables random access and parallel processing.
/// It checks for edge cases like snappy-compressed files and stdin input.
///
/// # Arguments
///
/// * `path` - Path to the CSV file to index
/// * `rconfig` - CSV reader configuration
///
/// # Returns
///
/// * `Ok(())` - Index was successfully created
/// * `Err(CliError)` - If index creation failed
pub fn create_index_for_file(path: &Path, rconfig: &Config) -> CliResult<()> {
    // Don't create index for snappy-compressed files
    if path.to_string_lossy().to_ascii_lowercase().ends_with(".sz") {
        return fail_clierror!(
            "Cannot create index for snappy-compressed files. Please decompress first."
        );
    }

    let pidx = idx_path(path);
    log::info!("Auto-creating index file: {}", pidx.display());

    let mut rdr = rconfig.reader_file()?;
    let idxfile = fs::File::create(&pidx)?;
    let mut wtr = BufWriter::with_capacity(DEFAULT_WTR_BUFFER_CAPACITY, idxfile);

    RandomAccessSimple::create(&mut rdr, &mut wtr)?;
    wtr.flush()?;

    log::info!("Successfully created index file: {}", pidx.display());
    Ok(())
}

/// Samples records from a CSV file for memory estimation.
///
/// Reads the first `sample_size` records for a quick estimate of average record size.
/// First-N sampling is intentionally chosen over reservoir sampling here because this
/// function is called before the main processing loop — a full-file scan would negate
/// the benefit of the estimate.
///
/// # Arguments
///
/// * `rconfig` - CSV reader configuration
/// * `sample_size` - Number of records to sample (typically 1000)
///
/// # Returns
///
/// * `Some(Vec<ByteRecord>)` - Sample records if available
/// * `None` - If no records could be sampled
pub fn sample_records(rconfig: &Config, sample_size: usize) -> Option<Vec<ByteRecord>> {
    let mut samples = Vec::with_capacity(sample_size);
    if let Ok(mut sample_rdr) = rconfig.reader() {
        let _ = sample_rdr.byte_headers(); // consume header row
        for (i, record_result) in sample_rdr.byte_records().enumerate() {
            if i >= sample_size {
                break;
            }
            if let Ok(record) = record_result {
                samples.push(record);
            } else {
                break;
            }
        }
    }
    if samples.is_empty() {
        None
    } else {
        Some(samples)
    }
}

/// Calculates dynamic chunk size based on available memory and record sampling.
///
/// This function estimates an appropriate chunk size by:
/// 1. Calculating average record size from sample records
/// 2. Getting available system memory
/// 3. Calculating memory per chunk (80% of available memory / number of jobs)
/// 4. Dividing memory per chunk by average record size
///
/// # Arguments
///
/// * `idx_count` - Total number of records in the file
/// * `njobs` - Number of parallel jobs
/// * `sample_records` - Optional slice of sample records for memory estimation
/// * `estimate_fn` - Function to estimate memory usage per record
///
/// # Returns
///
/// Calculated chunk size (number of records per chunk)
pub fn calculate_dynamic_chunk_size<F>(
    idx_count: u64,
    njobs: usize,
    sample_records: Option<&[ByteRecord]>,
    estimate_fn: F,
) -> usize
where
    F: Fn(&ByteRecord) -> usize,
{
    if let Some(samples) = sample_records {
        if samples.is_empty() {
            // No samples available, fall back to CPU-based chunking
            log::warn!("No records available for sampling, falling back to CPU-based chunking");
            return chunk_size(idx_count as usize, njobs);
        }

        let total_size: usize = samples.iter().map(&estimate_fn).sum();

        // samples.len() is guaranteed to be positive here as we checked above that it is not empty
        let avg_record_size = (total_size / samples.len()).max(1024);

        // Get available memory from system
        let mut sys = System::new();
        sys.refresh_memory();
        let avail_mem = sys.available_memory();

        // Calculate chunk size based on available memory
        // Use 80% of available memory divided by number of jobs
        #[allow(clippy::cast_precision_loss)]
        let memory_per_chunk = ((avail_mem as f64 * SAFETY_MARGIN) / njobs as f64) as usize;
        debug_assert!(avg_record_size > 0, "avg_record_size must be positive");
        let memory_based_chunk_size = memory_per_chunk / avg_record_size.max(1);

        // Ensure chunk size is reasonable
        let memory_based_chunk_size = memory_based_chunk_size.max(1).min(idx_count as usize);

        // Calculate CPU-based chunk size for optimal parallelization
        let cpu_based_chunk_size = chunk_size(idx_count as usize, njobs);

        // Calculate how many chunks each approach would create
        let memory_based_chunks = num_of_chunks(idx_count as usize, memory_based_chunk_size);
        let cpu_based_chunks = num_of_chunks(idx_count as usize, cpu_based_chunk_size);

        // Prefer CPU-based chunking if:
        // 1. Memory-based chunk size is smaller than CPU-based (degenerate/low memory), OR
        // 2. It creates more chunks (better parallelization)
        if memory_based_chunk_size <= cpu_based_chunk_size || cpu_based_chunks > memory_based_chunks
        {
            cpu_based_chunk_size
        } else {
            memory_based_chunk_size
        }
    } else {
        // No sample records provided, fall back to CPU-based chunking
        chunk_size(idx_count as usize, njobs)
    }
}

pub type Idx = Option<usize>;

pub fn range(start: Idx, end: Idx, len: Idx, index: Idx) -> Result<(usize, usize), String> {
    match (start, end, len, index) {
        (None, None, None, Some(i)) => Ok((i, i + 1)),
        (_, _, _, Some(_)) => fail!("--index cannot be used with --start, --end or --len"),
        (_, Some(_), Some(_), None) => {
            fail!("--end and --len cannot be used at the same time.")
        },
        (_, None, None, None) => Ok((start.unwrap_or(0), usize::MAX)),
        (_, Some(e), None, None) => {
            let s = start.unwrap_or(0);
            if s > e {
                fail_format!(
                    "The end of the range ({e}) must be greater than or\nequal to the start of \
                     the range ({s})."
                )
            } else {
                Ok((s, e))
            }
        },
        (_, None, Some(l), None) => {
            let s = start.unwrap_or(0);
            Ok((s, s + l))
        },
    }
}

/// Represents a filename template of the form `"{}.csv"`, where `"{}"` is
/// the place to insert the part of the filename generated by `qsv`.
#[cfg(any(feature = "feature_capable", feature = "lite"))]
#[derive(Clone)]
pub struct FilenameTemplate {
    prefix: String,
    suffix: String,
}

#[cfg(any(feature = "feature_capable", feature = "lite"))]
impl FilenameTemplate {
    /// Generate a new filename using `unique_value` to replace the `"{}"`
    /// in the template.
    pub fn filename(&self, unique_value: &str) -> String {
        format!("{}{unique_value}{}", self.prefix, self.suffix)
    }

    /// Create a new, writable file in directory `path` with a filename
    /// using `unique_value` to replace the `"{}"` in the template.  Note
    /// that we do not output headers; the caller must do that if
    /// desired.
    pub fn writer<P>(
        &self,
        path: P,
        unique_value: &str,
    ) -> std::io::Result<csv::Writer<Box<dyn std::io::Write + 'static>>>
    where
        P: AsRef<Path>,
    {
        let filename = self.filename(unique_value);
        let full_path = path.as_ref().join(filename);
        if let Some(parent) = full_path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        let spath = Some(full_path.display().to_string());
        Config::new(spath.as_ref()).writer()
    }
}

#[cfg(any(feature = "feature_capable", feature = "lite"))]
impl<'de> Deserialize<'de> for FilenameTemplate {
    fn deserialize<D: Deserializer<'de>>(d: D) -> Result<FilenameTemplate, D::Error> {
        let raw = String::deserialize(d)?;
        let chunks = raw.split("{}").collect::<Vec<_>>();
        if chunks.len() == 2 {
            Ok(FilenameTemplate {
                prefix: chunks[0].to_owned(),
                suffix: chunks[1].to_owned(),
            })
        } else {
            Err(D::Error::custom(
                "The --filename argument must contain one '{}'.",
            ))
        }
    }
}

pub fn init_logger() -> CliResult<(String, flexi_logger::LoggerHandle)> {
    use flexi_logger::{Cleanup, Criterion, FileSpec, Logger, Naming};

    let qsv_log_env = env::var("QSV_LOG_LEVEL").unwrap_or_else(|_| {
        #[cfg(feature = "qsvmcp")]
        {
            "info".to_string()
        }
        #[cfg(not(feature = "qsvmcp"))]
        {
            "off".to_string()
        }
    });
    let qsv_log_dir = env::var("QSV_LOG_DIR").unwrap_or_else(|_| ".".to_string());
    let write_mode = if get_envvar_flag("QSV_LOG_UNBUFFERED") {
        flexi_logger::WriteMode::Direct
    } else {
        flexi_logger::WriteMode::BufferAndFlush
    };

    let logger_handle = Logger::try_with_env_or_str(qsv_log_env);

    match logger_handle {
        Ok(logger) => {
            let logger = logger
                .use_utc()
                .log_to_file(
                    FileSpec::default()
                        .directory(qsv_log_dir)
                        .suppress_timestamp(),
                )
                .write_mode(write_mode)
                .format_for_files(flexi_logger::detailed_format)
                .o_append(true)
                .rotate(
                    Criterion::Size(20_000_000), // 20 mb
                    Naming::Numbers,
                    Cleanup::KeepLogAndCompressedFiles(10, 100),
                )
                .start()?;
            let qsv_args: String = if log_enabled!(log::Level::Info) {
                env::args().skip(1).collect::<Vec<_>>().join(" ")
            } else {
                String::new()
            };
            log::info!("START: {qsv_args}");
            Ok((qsv_args, logger))
        },
        Err(e) => Err(CliError::Other(format!("Failed to initialize logger: {e}"))),
    }
}

/// Turn a `self_update` error into an accurate, actionable message.
///
/// rc.4's `Error` is `#[non_exhaustive]` with structured HTTP variants, so we can
/// distinguish a genuine GitHub rate-limit / auth failure (403/401/429) from a network
/// timeout or connection failure - instead of the old blanket "Github is rate-limiting"
/// message that masked every failure mode (including the fixed pagination-URL 403 bug).
#[cfg(feature = "self_update")]
fn describe_self_update_error(e: &self_update::errors::Error) -> String {
    use self_update::errors::Error;

    match e {
        // 429 (and 403 below) are how GitHub signals secondary/primary rate limits
        Error::HttpStatus { status: 429, .. } => {
            format!(
                "GitHub is rate-limiting self-update checks (HTTP 429). Try again in an hour: {e}"
            )
        },
        Error::Unauthorized { status: 403, .. } => {
            format!("GitHub returned HTTP 403 - likely rate-limiting. Try again in an hour: {e}")
        },
        Error::Unauthorized { status, .. } => {
            format!("GitHub authorization failed (HTTP {status}): {e}")
        },
        Error::HttpStatus { status, .. } => format!("GitHub returned HTTP {status}: {e}"),
        Error::NotFound { .. } => format!("Release not found on GitHub: {e}"),
        Error::Transport(_) => {
            format!("Network error reaching GitHub (timeout or connection failure): {e}")
        },
        _ => format!("Self-update check failed: {e}"),
    }
}

#[cfg(feature = "self_update")]
pub fn qsv_check_for_update(check_only: bool, no_confirm: bool) -> Result<bool, String> {
    use self_update::cargo_crate_version;

    if get_envvar_flag("QSV_NO_UPDATE") {
        return Ok(false);
    }

    let bin_name = match current_exe() {
        Ok(pb) => {
            if let Some(fs) = pb.file_stem() {
                fs.to_string_lossy().into_owned()
            } else {
                return fail!("Can't get the exec stem name");
            }
        },
        Err(e) => return fail_format!("Can't get the exec path: {e}"),
    };

    winfo!("Checking GitHub for updates...");

    let curr_version = cargo_crate_version!();
    let releases = match self_update::backends::github::ReleaseList::configure()
        .repo_owner("dathere")
        .repo_name("qsv")
        // the release listing is a small JSON response, so bound it with a short timeout
        // to avoid hanging indefinitely on a stalled connection. Pair with a couple of
        // retries (exponential backoff, per self_update defaults) for transient failures.
        .timeout(Duration::from_secs(20))
        .retries(2)
        .build()
    {
        Ok(releases_list) => match releases_list.fetch() {
            Ok(releases) => releases,
            Err(e) => {
                return fail!(describe_self_update_error(&e));
            },
        },
        Err(e) => {
            return fail!(describe_self_update_error(&e));
        },
    };
    let Some(first_release) = releases.latest() else {
        return fail!("No releases found on GitHub.");
    };
    let latest_release = first_release.version();

    log::info!("Current version: {curr_version} Latest Release: {latest_release}");

    let mut updated = false;
    let Ok(latest_release_sv) = semver::Version::parse(latest_release) else {
        return fail_format!("Can't parse latest release version: {latest_release}");
    };
    let Ok(curr_version_sv) = semver::Version::parse(curr_version) else {
        return fail_format!("Can't parse current version: {curr_version}");
    };

    if latest_release_sv > curr_version_sv {
        eprintln!("Update {latest_release} available. Current version is {curr_version}.");
        eprintln!("Release notes: https://github.com/dathere/qsv/releases/tag/{latest_release}\n");
        if QSV_KIND.starts_with("prebuilt") && !check_only {
            match self_update::backends::github::Update::configure()
                .repo_owner("dathere")
                .repo_name("qsv")
                .bin_name(&bin_name)
                .show_download_progress(true)
                .show_output(false)
                .no_confirm(no_confirm)
                .current_version(curr_version)
                .verifying_keys([*include_bytes!("qsv-zipsign-public.key")])
                // NOTE: on the Update builder, timeout() also bounds the multi-MB binary
                // download (a total-request timeout), so use a generous cap here - NOT the
                // short listing timeout - to avoid aborting legitimate slow downloads.
                // retries() only re-attempts before any bytes are streamed to disk.
                .timeout(Duration::from_secs(600))
                .retries(2)
                .build()
            {
                Ok(update_job) => match update_job.update() {
                    Ok(status) => {
                        updated = true;
                        let update_status = format!(
                            "Update successful for {}: `{}`!",
                            bin_name,
                            status.version()
                        );
                        winfo!("{update_status}");
                    },
                    Err(e) => werr!("Update job error: {}", describe_self_update_error(&e)),
                },
                Err(e) => werr!("Update builder error: {}", describe_self_update_error(&e)),
            }
        } else if check_only {
            winfo!("Use the --update option to upgrade {bin_name} to the latest release.");
        } else {
            // we don't want to overwrite manually curated/configured qsv installations.
            // If QSV_KIND is not "prebuilt", just inform the user of the new release, and let them
            // rebuild their qsvs the way they like it, instead of overwriting it with
            // our prebuilt binaries.
            winfo!(
                r"This qsv was {QSV_KIND}. self-update does not work for manually {QSV_KIND} binaries.
If you wish to update to the latest version of qsv, manually install/compile from source.
Self-update only works with prebuilt binaries released on GitHub https://github.com/dathere/qsv/releases/latest"
            );
        }
    } else {
        winfo!("Up to date ({curr_version})... no update required.");
    }

    if !check_only
        && let Ok(status_code) =
            send_hwsurvey(&bin_name, updated, latest_release, curr_version, false)
    {
        log::info!("HW survey sent. Status code: {status_code}");
    }

    Ok(updated)
}

#[cfg(not(feature = "self_update"))]
pub fn qsv_check_for_update(_check_only: bool, _no_confirm: bool) -> Result<bool, String> {
    Err("Self-update is disabled in this build.".to_string())
}

// how often a background (non-flag) update check is allowed to hit GitHub: ~twice a month
#[cfg(feature = "self_update")]
const UPDATE_CHECK_INTERVAL: Duration = Duration::from_hours(360);

/// Background update check, throttled to at most twice a month via a persisted stamp file.
///
/// Used by the no-command screen and the `help` command so they don't hit GitHub on every
/// invocation. The explicit `--update`/`--updatenow` flags bypass this and call
/// `qsv_check_for_update` directly. Best-effort: all errors are swallowed.
#[cfg(feature = "self_update")]
pub fn qsv_check_for_update_throttled() {
    // honor the opt-out BEFORE touching the filesystem
    if get_envvar_flag("QSV_NO_UPDATE") {
        return;
    }

    // stamp lives in the qsv cache dir: QSV_CACHE_DIR override, else ~/.qsv-cache
    let raw = std::env::var("QSV_CACHE_DIR")
        .ok()
        .filter(|s| !s.is_empty())
        .unwrap_or_else(|| "~/.qsv-cache".to_string());
    let Some(cache_dir) = expand_tilde(&raw) else {
        return; // can't resolve home; skip silently
    };

    // create the parent dir so record_check() can persist. If it can't, should_check()
    // would return Ok(true) forever => a check on EVERY startup (worse than no throttle).
    if fs::create_dir_all(&cache_dir).is_err() {
        return;
    }

    let stamp_path = cache_dir.join(".last_update_check");
    let guard =
        self_update::check_interval::UpdateCheckGuard::new(stamp_path, UPDATE_CHECK_INTERVAL);

    // on a genuine IO error reading the stamp, DON'T check (avoid hammering GitHub);
    // a missing/corrupt stamp already yields Ok(true), so first run still checks.
    if guard.should_check().unwrap_or(false) {
        _ = qsv_check_for_update(true, false); // best-effort; ignore network errors
        _ = guard.record_check(); // record regardless of outcome => real throttle
    }
}

#[cfg(not(feature = "self_update"))]
pub fn qsv_check_for_update_throttled() {}

// the qsv hwsurvey allows us to keep a better
// track of qsv's usage in the wild, so we can do a
// better job of prioritizing platforms/features we support
// no personally identifiable information is collected
#[cfg(feature = "self_update")]
fn send_hwsurvey(
    bin_name: &str,
    updated: bool,
    latest_release: &str,
    curr_version: &str,
    dry_run: bool,
) -> Result<reqwest::StatusCode, String> {
    use serde_json::json;

    static HW_SURVEY_URL: &str =
        "https://4dhmneehnl.execute-api.us-east-1.amazonaws.com/dev/qsv-hwsurvey";

    let mut sys = System::new();
    sys.refresh_all();
    let total_mem = sys.total_memory();
    let kernel_version =
        sysinfo::System::kernel_version().unwrap_or_else(|| "Unknown kernel".to_string());
    let long_os_version =
        sysinfo::System::long_os_version().unwrap_or_else(|| "Unknown OS version".to_string());
    let cpu_count = sys.cpus().len();
    let physical_cpu_count = sysinfo::System::physical_core_count().unwrap_or_default();
    let cpu_vendor_id = sys.cpus()[0].vendor_id();
    let cpu_brand = sys.cpus()[0].brand().trim();
    let cpu_freq = sys.cpus()[0].frequency();
    let long_id: u128 = std::time::SystemTime::now()
        .duration_since(std::time::SystemTime::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis();
    // the id doubles as a timestamp
    // we first get number of milliseconds since UNIX EPOCH
    // and then cast to u64 as serde_json cannot serialize u128
    let id: u64 = long_id.try_into().unwrap_or_default();
    let hwsurvey_json = json!(
        {
            "id": id,
            "variant": bin_name,
            "kind": QSV_KIND,
            "ver": if updated { latest_release } else { curr_version },
            "updated": updated,
            "prev_ver": curr_version,
            "cpu_phy_cores": physical_cpu_count,
            "cpu_log_cores": cpu_count,
            "cpu_vendor": cpu_vendor_id,
            "cpu_brand": cpu_brand,
            "cpu_freq": cpu_freq,
            "mem": total_mem,
            "kernel": kernel_version,
            "os": long_os_version,
            "target": TARGET,
        }
    );
    log::debug!("hwsurvey: {hwsurvey_json}");

    let mut survey_done = false;
    let mut status = reqwest::StatusCode::OK;
    if dry_run {
        log::info!("Survey dry run. hw survey compiled successfully, but not sent.");
    } else {
        let client = match create_reqwest_blocking_client(
            Some(default_user_agent()),
            30, // default timeout for hw survey
            None,
        ) {
            Ok(c) => c,
            Err(e) => return fail_format!("Cannot build hw_survey reqwest client: {e}"),
        };

        match client
            .post(HW_SURVEY_URL)
            .body(hwsurvey_json.to_string())
            .header(reqwest::header::CONTENT_TYPE, "application/json")
            .header(reqwest::header::HOST, "qsv.rs")
            .send()
        {
            Ok(resp) => {
                log::debug!("hw_survey response sent: {resp:?}");
                status = resp.status();
                survey_done = status.is_success();
            },
            Err(e) => {
                log::warn!("Cannot send hw survey: {e}");
                status = reqwest::StatusCode::BAD_REQUEST;
            },
        }
    }
    if survey_done || dry_run {
        Ok(status)
    } else {
        fail!("hw survey failed.")
    }
}

pub fn safe_header_names(
    headers: &csv::StringRecord,
    check_first_char: bool,
    conditional: bool,
    reserved_names: Option<&Vec<String>>,
    unsafe_prefix: &str,
    keep_case: bool,
    collapse: bool,
    unicode: bool,
) -> (Vec<String>, u16) {
    // Create "safe" var/key names - to support dynfmt/url-template, valid python vars & db-safe
    // column names. Fold to lowercase if keep_case is false. Trim leading & trailing whitespace.
    // Replace whitespace/non-alphanumeric) with _. If name starts with a number & check_first_char
    // is true, prepend the unsafe_prefix. If a column with the same name already exists,
    // append a sequence suffix (e.g. _n). Names are limited to 60 BYTES in length (snapped to
    // a UTF-8 char boundary), matching `is_safe_name`'s byte check and staying within
    // Postgres' default NAMEDATALEN of 63 bytes — including any disambiguation suffix.
    // Empty names are replaced with unsafe_prefix as well.
    //
    // If `collapse` is true, runs of non-alphanumeric characters collapse into a
    // single `_` (the regex gains a `+`); since `_` is itself non-alphanumeric,
    // runs of literal underscores collapse too. If `unicode` is true, unicode
    // letters & numbers are preserved instead of being stripped to ASCII.

    // If conditional = true & reserved_names is none, only rename the header if its not safe
    let prefix = if unsafe_prefix.is_empty() {
        "_"
    } else {
        unsafe_prefix
    };
    // Select the sanitization regex based on the (collapse, unicode) axes. Each
    // `regex_oncelock!` literal compiles to its own static OnceLock; we just pick
    // the reference at runtime. Appending `+` collapses runs into one `_`.
    let safename_regex = match (collapse, unicode) {
        (false, false) => regex_oncelock!(r"[^A-Za-z0-9]"),
        (true, false) => regex_oncelock!(r"[^A-Za-z0-9]+"),
        (false, true) => regex_oncelock!(r"[^\p{L}\p{N}]"),
        (true, true) => regex_oncelock!(r"[^\p{L}\p{N}]+"),
    };
    let mut changed_count = 0_u16;
    let mut name_vec: Vec<String> = Vec::with_capacity(headers.len());
    let mut safe_name: String;
    let mut safename_always: String;
    let mut safename_candidate: String;
    let mut final_candidate: String;
    let mut buf_wrk = String::new();

    for header_name in headers {
        let reserved_found = if let Some(reserved_names_vec) = reserved_names {
            if keep_case {
                header_name.clone_into(&mut buf_wrk);
            } else {
                to_lowercase_into(header_name, &mut buf_wrk);
            }
            reserved_names_vec
                .iter()
                .any(|reserved_name| reserved_name == &buf_wrk)
        } else {
            false
        };
        safe_name = if conditional
            && is_conditionally_safe(header_name, collapse, unicode)
            && !reserved_found
        {
            header_name.to_string()
        } else {
            // Trim first so whitespace-only headers are treated as empty —
            // the regex replace below would otherwise yield "" and the
            // as_bytes()[0] access would panic.
            safename_always = if header_name.trim().is_empty() {
                prefix.to_string()
            } else {
                safename_regex
                    .replace_all(header_name.trim(), "_")
                    .to_string()
            };
            // Check the first *scalar value*, not the first byte: in unicode
            // mode a preserved leading Unicode digit (e.g. "２col", "٣col") must
            // also trigger the unsafe prefix, otherwise it bypasses the
            // "starts with a number" safety rule. In ASCII mode the string is
            // pure ASCII, so is_ascii_digit on the first char is equivalent to
            // the previous first-byte check.
            if check_first_char
                && safename_always.chars().next().is_some_and(|c| {
                    if unicode {
                        c.is_numeric()
                    } else {
                        c.is_ascii_digit()
                    }
                })
            {
                safename_always = format!("{prefix}{safename_always}");
            }

            safename_candidate = if reserved_found {
                log::warn!("\"{safename_always}\" is a reserved name: {reserved_names:?}");
                format!("reserved_{safename_always}")
            } else {
                safename_always
            };

            // Apply lowercase and the "_<prefix>" prepend BEFORE truncation —
            // case-folding can change byte length (e.g. some Unicode glyphs
            // expand on lowercase) and prepending the prefix adds bytes too,
            // so doing them post-truncation could push the final name past
            // the 60-byte budget. Order: lowercase → leading-_ prefix → byte
            // truncate at a char boundary.
            final_candidate = if keep_case {
                safename_candidate
            } else {
                safename_candidate.to_lowercase()
            };
            if prefix != "_" && final_candidate.starts_with('_') {
                final_candidate = format!("{prefix}{final_candidate}");
            }
            if final_candidate.len() > 60 {
                final_candidate.truncate(final_candidate.floor_char_boundary(60));
            }
            final_candidate
        };
        let mut sequence_suffix = 2_u16;
        let mut candidate_name = safe_name.clone();
        while name_vec.contains(&candidate_name) {
            // Trim the base so that base + "_<n>" stays within the 60-byte
            // limit; otherwise the pre-truncated `safe_name` plus suffix could
            // overflow (e.g. safe_name=60 bytes + "_2" → 62 bytes), pushing
            // past Postgres' default NAMEDATALEN of 63 bytes and out of sync
            // with `is_safe_name`'s byte-based length check. We measure in
            // bytes (not chars) and snap to a UTF-8 char boundary so non-ASCII
            // headers can't smuggle in a 240-byte name behind a 60-char count.
            // Note: when two distinct long source headers share the same
            // first `base_max` bytes, they'll share the suffix counter — i.e.
            // the second header's first occurrence may be `_3` rather than
            // `_2`. This is intentional: the trimmed prefix is the actual
            // identifier downstream consumers see, so collisions in that
            // namespace must be disambiguated, not the original header's.
            let suffix = format!("_{sequence_suffix}");
            let base_max = 60_usize.saturating_sub(suffix.len());
            let base = if safe_name.len() > base_max {
                &safe_name[..safe_name.floor_char_boundary(base_max)]
            } else {
                safe_name.as_str()
            };
            candidate_name = format!("{base}{suffix}");
            sequence_suffix += 1;
        }
        if candidate_name.ne(header_name) {
            changed_count += 1;
        }
        name_vec.push(candidate_name);
    }
    log::debug!("safe header names: {name_vec:?}");
    (name_vec, changed_count)
}

#[inline]
pub fn is_safe_name(header_name: &str) -> bool {
    if header_name.trim().is_empty()
        || header_name.trim_start_matches('_').is_empty()
        || header_name.len() > 60
    {
        return false;
    }
    let first_character = header_name.trim_start_matches('_').as_bytes()[0];
    if first_character.is_ascii_digit() || first_character.is_ascii_whitespace() {
        return false;
    }
    let safename_re = regex_oncelock!(r"^[\w\-\s]+$");
    safename_re.is_match(header_name)
}

/// Whether `header_name` may be kept as-is in conditional mode given the active
/// `collapse`/`unicode` options. Beyond `is_safe_name`, a name is NOT
/// preservable when:
///  - `collapse` is set and it contains a run of 2+ characters the active rewrite would collapse
///    (e.g. the "__" in `a__b`); or
///  - `unicode` is set and it contains a character the unicode sanitizer (`[^\p{L}\p{N}]`) would
///    replace but `is_safe_name`'s lenient `\w` admits — e.g. combining marks (`\p{M}`) or
///    connector punctuation — apart from the quoted-identifier separators (`_`, `-`, whitespace);
///    or
///  - `unicode` is set and it starts with a Unicode digit (the rewrite prefixes it).
///
/// This keeps conditional mode consistent with always/verify under those flags —
/// e.g. `--mode c --collapse` rewrites `a__b` to `a_b` and `--mode c --unicode`
/// rewrites a decomposed `cafe\u{301}` to `cafe_`, while still preserving genuine
/// quoted identifiers like `Col with Embedded Spaces` and composed unicode names.
#[inline]
fn is_conditionally_safe(header_name: &str, collapse: bool, unicode: bool) -> bool {
    if !is_safe_name(header_name) {
        return false;
    }
    if collapse {
        let run_re = if unicode {
            regex_oncelock!(r"[^\p{L}\p{N}]{2,}")
        } else {
            regex_oncelock!(r"[^A-Za-z0-9]{2,}")
        };
        if run_re.is_match(header_name.trim()) {
            return false;
        }
    }
    if unicode {
        // Reject any character the unicode sanitizer would replace, except the
        // quoted-identifier separators we intentionally preserve (`_`, `-`,
        // whitespace). Without this, `\w`-only chars such as combining marks
        // would be kept here yet rewritten by always/verify.
        let unicode_extra_re = regex_oncelock!(r"[^\p{L}\p{N}_\-\s]");
        if unicode_extra_re.is_match(header_name) {
            return false;
        }
        // A preserved leading Unicode digit must also be rejected so the
        // "starts with a number" rule still applies (e.g. `２col`).
        if header_name
            .trim_start_matches('_')
            .chars()
            .next()
            .is_some_and(char::is_numeric)
        {
            return false;
        }
    }
    true
}

pub fn log_end(mut qsv_args: String, now: std::time::Instant) {
    #[cfg(feature = "polars")]
    use crate::config::TEMP_FILE_DIR;

    #[cfg(feature = "polars")]
    if let Some(temp_dir) = TEMP_FILE_DIR.get() {
        // if polars is enabled, we need to remove the temporary directory
        // after the command finishes. This is using unwrap_or_default()
        // to avoid panics if the directory is already deleted.
        std::fs::remove_dir_all(temp_dir).unwrap_or_default();
    }
    if log::log_enabled!(log::Level::Info) {
        let ellipsis = if qsv_args.len() > 24 {
            utf8_truncate(&mut qsv_args, 24);
            "..."
        } else {
            ""
        };
        log::info!(
            "END \"{qsv_args}{ellipsis}\" elapsed: {}",
            now.elapsed().as_secs_f32()
        );
    }
}

/// Truncates a UTF-8 encoded string to fewer than `maxsize` bytes, preserving
/// valid UTF-8.
///
/// The result is at most `maxsize - 1` bytes (or 0 if `maxsize == 0`). The
/// "off-by-one" is intentional: stats `*PREVIEW*` snapshots are calibrated to
/// it. Don't change the offset without re-blessing those tests.
///
/// # Caller convention
///
/// To produce a string of at most `N` bytes (e.g., before appending an
/// ellipsis), callers should:
///
/// ```ignore
/// if input.len() > N {
///     util::utf8_truncate(&mut input, N + 1);  // +1 cancels the internal -1
///     input.push_str("...");
/// }
/// ```
///
/// The outer length guard avoids touching strings that already fit; the
/// `N + 1` argument compensates for the internal `maxsize - 1` so the result
/// is exactly `N` bytes before the suffix is appended. See the call site in
/// `src/cmd/stats.rs` (`antimodes` truncation).
///
/// # Arguments
///
/// * `input` - A mutable reference to the String to truncate
/// * `maxsize` - One past the maximum byte length the result may occupy
///
/// Uses `str::floor_char_boundary` so multi-byte characters are never split.
pub fn utf8_truncate(input: &mut String, maxsize: usize) {
    if input.len() > maxsize {
        // Find the largest char boundary strictly less than maxsize.
        // We deliberately use maxsize - 1 (not maxsize) because callers and
        // snapshot tests rely on truncation cutting one byte short of the
        // requested maximum (used by stats *PREVIEW* truncation).
        input.truncate(if maxsize > 0 {
            input.floor_char_boundary(maxsize - 1)
        } else {
            0
        });
    }
}

#[test]
#[cfg(feature = "self_update")]
fn test_hw_survey() {
    // we have this test primarily to exercise the sysinfo module
    assert!(send_hwsurvey("qsv", false, "0.0.2", "0.0.1", true).is_ok());
}

pub struct ColumnNameParser {
    chars: Vec<char>,
    pos:   usize,
}

impl ColumnNameParser {
    pub fn new(s: &str) -> ColumnNameParser {
        ColumnNameParser {
            chars: s.chars().collect(),
            pos:   0,
        }
    }

    pub fn parse(&mut self) -> Result<Vec<String>, String> {
        let mut new_cols_name = vec![];
        loop {
            if self.cur().is_none() {
                break;
            }
            if self.cur() == Some('"') {
                self.bump();
                new_cols_name.push(self.parse_quoted_name()?);
            } else {
                new_cols_name.push(self.parse_name());
            }
            self.bump();
        }
        Ok(new_cols_name)
    }

    fn cur(&self) -> Option<char> {
        self.chars.get(self.pos).copied()
    }

    const fn bump(&mut self) {
        if self.pos < self.chars.len() {
            self.pos += 1;
        }
    }

    fn is_end_of_field(&self) -> bool {
        self.cur().is_none_or(|c| c == ',')
    }

    fn parse_quoted_name(&mut self) -> Result<String, String> {
        let mut name = String::new();
        loop {
            match self.cur() {
                None => {
                    return fail!("Unclosed quote, missing \".");
                },
                Some('"') => {
                    self.bump();
                    if self.cur() == Some('"') {
                        self.bump();
                        name.push('"');
                        name.push('"');
                        continue;
                    }
                    break;
                },
                Some(c) => {
                    name.push(c);
                    self.bump();
                },
            }
        }
        Ok(name)
    }

    fn parse_name(&mut self) -> String {
        let mut name = String::new();
        loop {
            if self.is_end_of_field() {
                break;
            }
            // safety: we know that the cur() will not be None as we checked above
            name.push(self.cur().unwrap());
            self.bump();
        }
        name
    }
}

#[inline]
/// Rounds a floating point number to a specified number of decimal places.
///
/// This function takes a 64-bit floating point number and rounds it to the specified number of
/// decimal places using "Bankers Rounding" (Midpoint Nearest Even) strategy. It returns the result
/// as a String.
///
/// # Arguments
///
/// * `dec_f64` - The floating point number to round
/// * `places` - The number of decimal places to round to. If set to 9999, no rounding is performed.
///
/// # Returns
///
/// * A String containing the rounded number with trailing zeros removed and -0.0 normalized to 0.0
pub fn round_num(dec_f64: f64, places: u32) -> String {
    use rust_decimal::{Decimal, RoundingStrategy};

    if dec_f64.is_nan() {
        return String::new();
    }

    // if places is the sentinel value 9999, we don't round, just return the number as is
    if places == 9999 {
        return zmij::Buffer::new().format(dec_f64).to_owned();
    }

    // use from_f64_retain, so we have all the excess bits before rounding with
    // round_dp_with_strategy as from_f64 will prematurely round when it drops the excess bits
    let Some(dec_num) = Decimal::from_f64_retain(dec_f64) else {
        return String::new();
    };

    // round using Midpoint Nearest Even Rounding Strategy AKA "Bankers Rounding."
    // https://docs.rs/rust_decimal/latest/rust_decimal/enum.RoundingStrategy.html#variant.MidpointNearestEven
    // we also normalize to remove trailing zeroes and to change -0.0 to 0.0.
    dec_num
        .round_dp_with_strategy(places, RoundingStrategy::MidpointNearestEven)
        .normalize()
        .to_string()
}

#[inline]
/// Transforms a byte slice into a `ByteString` with optional case-insensitive conversion.
///
/// This function takes a byte slice and attempts to convert it to a UTF-8 string. If successful,
/// it trims whitespace and optionally converts to lowercase. If the input is not valid UTF-8,
/// it returns the original bytes unchanged.
///
/// It's fine-tuned for speed and memory usage, using simdutf8 for UTF-8 validation and
/// `to_lowercase_into` for non-allocating, in-place lowercase conversion.
///
/// # Arguments
///
/// * `bs` - The input byte slice to transform
/// * `casei` - If true, converts the string to lowercase. If false, leaves case unchanged.
///
/// # Returns
///
/// * A `ByteString` (Vec<u8>) containing the transformed bytes
pub fn transform(bs: &[u8], casei: bool) -> ByteString {
    if let Ok(s) = simdutf8::basic::from_utf8(bs) {
        if casei {
            let mut buffer = String::with_capacity(bs.len());
            to_lowercase_into(s.trim(), &mut buffer);
            buffer.into_bytes()
        } else {
            s.trim().as_bytes().to_vec()
        }
    } else {
        bs.to_vec()
    }
}

pub fn load_dotenv() -> CliResult<()> {
    // First, check if there is a QSV_DOTENV_PATH environment variable set
    // if there is, use that as the .env file.
    // Second, use the default .env file in the current directory.
    // If there is no .env file in the current directory, check if there is
    // an .env file with the same filestem as the binary, in the same directory as the binary.
    // If there is, use that. Failing that, qsv proceeds with its default settings and
    // whatever manually set environment variables are present.

    if let Ok(dotenv_path) = std::env::var("QSV_DOTENV_PATH") {
        // <NONE> is a sentinel value to disable dotenv processing
        if dotenv_path == "<NONE>" {
            log::warn!("dotenv processing disabled with QSV_DOTENV_PATH=<NONE>");
            return Ok(());
        }

        let canonical_dotenv_path = std::fs::canonicalize(dotenv_path)?;
        if let Err(e) = dotenvy::from_filename_override(canonical_dotenv_path.clone()) {
            return fail_clierror!(
                "Cannot process .env file set in QSV_DOTENV_PATH - {}: {e}",
                canonical_dotenv_path.display()
            );
        }
        log::info!("Using .env file: {}", canonical_dotenv_path.display());
        return Ok(());
    }

    // check if there is an .env file in the current directory
    if dotenvy::dotenv_override().is_ok() {
        log::info!(
            "Using .env file in current directory: {}",
            std::env::current_dir()?.display()
        );
    } else {
        // no .env file in the current directory or it was invalid
        // now check if there is an .env file with the same name as the executable
        // in the same directory as the executable
        let qsv_binary_path = current_exe()?;

        let qsv_dir = qsv_binary_path.parent().ok_or("No parent directory")?;

        // safety: we know that the file_stem() will not be None as we checked above
        let qsv_binary_filestem = qsv_binary_path
            .file_stem()
            .unwrap()
            .to_str()
            .unwrap()
            .to_string();

        let mut qsv_binary_envprofile = qsv_dir.to_path_buf();
        qsv_binary_envprofile.set_file_name(format!("{qsv_binary_filestem}.env"));

        if std::path::Path::new(&qsv_binary_envprofile).exists() {
            log::info!(
                "Using binary .env file: {}",
                qsv_binary_envprofile.display()
            );
            if let Err(e) = dotenvy::from_filename_override(qsv_binary_envprofile.clone()) {
                return fail_clierror!(
                    "Cannot process binary .env file - {}: {e}",
                    qsv_binary_envprofile.display()
                );
            }
        } else {
            // there is no binary .env file, just use the default settings
            // and whatever manually set environment variables are present
            log::info!(
                "No valid .env file found. Proceeding with default settings and current \
                 environment variable settings."
            );
        }
    }

    Ok(())
}

#[inline]
pub fn get_envvar_flag(key: &str) -> bool {
    use std::{
        collections::HashSet,
        sync::{Mutex, OnceLock},
    };

    static UNRECOGNIZED_WARNED: OnceLock<Mutex<HashSet<String>>> = OnceLock::new();

    let Ok(raw) = std::env::var(key) else {
        return false;
    };
    // Trim whitespace so values like " true" are recognized as truthy.
    let v = raw.trim().to_ascii_lowercase();
    let truthy = matches!(v.as_str(), "true" | "t" | "1" | "yes" | "y" | "on");
    let falsy = matches!(v.as_str(), "" | "false" | "f" | "0" | "no" | "n" | "off");
    if !truthy && !falsy {
        // Warn at most once per (key, value) pair across the process so callers
        // in hot loops don't spam the log with the same misconfiguration notice.
        // Decide whether to warn while holding the lock, then drop the guard
        // before calling into the logger — log::warn! can take its own locks
        // (flexi_logger), so we don't want to bridge them here.
        let warned = UNRECOGNIZED_WARNED.get_or_init(|| Mutex::new(HashSet::new()));
        let should_warn = match warned.lock() {
            Ok(mut set) => set.insert(format!("{key}={raw}")),
            Err(_) => false,
        };
        if should_warn {
            log::warn!("{key}={raw} is not a recognized boolean; treating as false");
        }
    }
    truthy
}

/// Returns `true` if `flag` appears in `argv` as either a standalone token
/// (`--flag`) or in the `--flag=value` form. Used by the OOM auto-fallback in
/// `stats` and `frequency` to detect whether the user explicitly passed an
/// option, since docopt fills in default values that are indistinguishable
/// from explicit user input on the parsed `Args` struct.
#[inline]
pub fn argv_has_flag(argv: &[&str], flag: &str) -> bool {
    let eq_prefix = format!("{flag}=");
    argv.iter().any(|a| *a == flag || a.starts_with(&eq_prefix))
}

/// Validates if a file is actually a Snappy-compressed file before attempting decompression
pub fn is_valid_snappy_file(path: &PathBuf) -> Result<bool, CliError> {
    let mut file = std::fs::File::open(path)?;
    let mut reader = BufReader::new(&mut file);

    // Try to create a FrameDecoder and read the first few bytes
    // This will fail immediately if the file doesn't have a valid Snappy header
    let decoder = snap::read::FrameDecoder::new(&mut reader);
    let mut buffer = Vec::with_capacity(50);

    match decoder.take(50).read_to_end(&mut buffer) {
        Ok(_) => {
            // Successfully read some bytes, this is likely a valid Snappy file
            log::debug!("File {} appears to be a valid Snappy file", path.display());
            Ok(true)
        },
        Err(e) => {
            // Failed to read, this is not a valid Snappy file
            log::debug!("File {} is not a valid Snappy file: {}", path.display(), e);
            Ok(false)
        },
    }
}

/// Decompresses a Snappy-compressed file to a temporary directory.
///
/// # Arguments
///
/// * `path` - Path to the file with `.sz` extension that should be decompressed
/// * `tmpdir` - Temporary directory where the decompressed file will be placed
///
/// # Returns
///
/// Returns the path to the decompressed file in the temporary directory as a `String`.
///
/// # Fallback Behavior
///
/// If a file has a `.sz` extension but is not actually a valid Snappy-compressed file,
/// this function falls back gracefully by copying the file to the temp directory as a
/// plain file (without the `.sz` extension). This prevents "corrupt input" errors when
/// plain CSV files are incorrectly detected as snappy-compressed (e.g., due to temp file
/// naming bugs or extension detection issues).
///
/// This fallback is necessary because this function is used by commands that require
/// file-based access (like `slice`, `lens`, `describegpt`, `tojsonl`, `joinp`, `sniff`)
/// which use `process_input()` to decompress files before creating a `Config`. Since
/// `process_input()` strips the `.sz` extension from the temp file path, `Config::io_reader()`
/// never sees the extension and cannot handle the fallback. Therefore, the validation and
/// fallback must happen here in `decompress_snappy_file()`.
///
/// Commands that use streaming (like `count`, `stats`, `frequency`) use `Config::io_reader()`
/// directly, which also has validation and fallback logic. Both code paths need their own
/// validation because they serve different purposes (file-based vs streaming access).
///
/// # Errors
///
/// Returns an error if:
/// - The file cannot be opened
/// - File validation fails (though this now falls back to plain file handling)
/// - The file cannot be copied to the temp directory (fallback case)
/// - Decompression fails for valid snappy files
pub fn decompress_snappy_file(
    path: &PathBuf,
    tmpdir: &tempfile::TempDir,
) -> Result<String, CliError> {
    // First, validate that this is actually a Snappy file
    if !is_valid_snappy_file(path)? {
        // File has .sz extension but is not a valid Snappy file.
        // Fall back gracefully by copying it to temp directory as a plain file.
        // This prevents "corrupt input" errors when plain CSV files are incorrectly
        // detected as snappy (e.g., due to temp file naming bugs).
        warn!(
            "File {} has .sz extension but is not a valid Snappy file. Treating as plain file.",
            path.display()
        );

        // Copy the file to temp directory with original name (without .sz)
        let file_stem = Path::new(&path)
            .file_stem()
            .and_then(|s| s.to_str())
            .unwrap_or("plain_file");
        let fallback_filepath = tmpdir.path().join(file_stem);
        std::fs::copy(path, &fallback_filepath)?;
        return Ok(format!("{}", fallback_filepath.display()));
    }

    // Proceed with decompression since we've validated the file
    let mut snappy_file = std::fs::File::open(path.clone())?;
    let mut snappy_reader = snap::read::FrameDecoder::new(&mut snappy_file);
    // safety: we know that file_stem() will not be None as we opened the file above
    let file_stem = Path::new(&path).file_stem().unwrap().to_string_lossy();
    let decompressed_filepath = tmpdir
        .path()
        .join(format!("qsv_temp_decompressed__{file_stem}"));
    let mut decompressed_file = std::fs::File::create(decompressed_filepath.clone())?;

    match std::io::copy(&mut snappy_reader, &mut decompressed_file) {
        Ok(num_bytes) => {
            decompressed_file.flush()?;
            // Explicitly close the write handle before returning. On Windows
            // the next reader to open this path can deadlock if a writer
            // handle is still alive in this process — observed as an
            // indefinite hang on Windows ARM64 (sniff URL → snappy → sniffer).
            drop(decompressed_file);
            log::debug!(
                "Successfully decompressed Snappy file: {} ({} bytes)",
                path.display(),
                num_bytes
            );
            Ok(format!("{}", decompressed_filepath.display()))
        },
        Err(e) => {
            // Clean up the partially created file
            let _ = std::fs::remove_file(&decompressed_filepath);
            fail_clierror!(
                "Failed to decompress Snappy file '{}': {}. The file may be corrupted or \
                 incomplete.",
                path.display(),
                e
            )
        },
    }
}

/// downloads a file from a url and saves it to a path
/// if `show_progress` is true, a progress bar will be shown
/// if `custom_user_agent` is Some, it will be used as the user agent
/// if `download_timeout` is Some, it will be used as the timeout in seconds. The
/// value is validated by `timeout_secs` and must be in the range 1..=3600;
/// values of 0 or > 3600 cause this function to return an error. If
/// `sample_size` is Some, it will be used as the number of bytes to download.
pub async fn download_file(
    url: &str,
    path: PathBuf,
    #[allow(unused_variables)] show_progress: bool,
    custom_user_agent: Option<String>,
    download_timeout: Option<u16>,
    sample_size: Option<u64>,
) -> CliResult<()> {
    use futures_util::StreamExt;

    let download_timeout = match download_timeout {
        Some(t) => std::time::Duration::from_secs(
            timeout_secs(t)
                .map_err(|e| CliError::Other(format!("Invalid download timeout: {e}")))?,
        ),
        None => std::time::Duration::from_secs(30),
    };

    // setup the reqwest client. create_reqwest_async_client calls
    // set_user_agent internally, so pass custom_user_agent through.
    let client = create_reqwest_async_client(
        custom_user_agent,
        download_timeout.as_secs() as u16,
        Some(url.to_string()),
    )?;

    let res = client.get(url).send().await?;

    // if we can't get the content length, set it to sentinel value
    let total_size = res.content_length().unwrap_or(u64::MAX);

    // progressbar setup
    #[cfg(any(feature = "feature_capable", feature = "lite"))]
    let show_progress = (show_progress || get_envvar_flag("QSV_PROGRESSBAR")) && total_size > 0;

    #[cfg(any(feature = "feature_capable", feature = "lite"))]
    let pb = ProgressBar::with_draw_target(Some(total_size), ProgressDrawTarget::stderr_with_hz(5));

    #[cfg(any(feature = "feature_capable", feature = "lite"))]
    if show_progress {
        pb.set_style(
            #[allow(clippy::to_string_in_format_args)]
            #[allow(clippy::literal_string_with_formatting_args)]
            ProgressStyle::default_bar()
                .template(if total_size == u64::MAX {
                    // only do a spinner if we don't know the total size
                    "{msg}\n{spinner:.green} ({bytes_per_sec})"
                } else {
                    "{msg}\n{spinner:.green} [{elapsed_precise}] [{wide_bar:.white/blue}] \
                     {bytes}/{total_bytes} ({bytes_per_sec}, {eta})"
                })
                .unwrap(),
        );
        pb.set_message(format!("Downloading {url}"));
    } else {
        pb.set_draw_target(ProgressDrawTarget::hidden());
    }

    let sample_size = sample_size.unwrap_or(0);

    // download chunks
    let mut file = BufWriter::with_capacity(DEFAULT_WTR_BUFFER_CAPACITY, File::create(path)?);
    let mut downloaded: u64 = 0;
    let mut stream = res.bytes_stream();

    while let Some(item) = stream.next().await {
        let chunk = item?;
        file.write_all(&chunk)?;
        let new = min(downloaded + (chunk.len() as u64), total_size);
        downloaded = new;

        #[cfg(any(feature = "feature_capable", feature = "lite"))]
        if show_progress {
            pb.set_position(new);
        }

        if sample_size > 0 && downloaded >= sample_size {
            break;
        }
    }

    #[cfg(any(feature = "feature_capable", feature = "lite"))]
    if show_progress {
        pb.finish_with_message(format!("Downloaded {url}"));
        eprintln!(); // newline after progress bar
    }

    Ok(file.flush()?)
}

/// this is a non-allocating `to_lowercase` that uses an existing buffer
/// and should be faster than the allocating `std::to_lowercase`
#[inline]
pub fn to_lowercase_into(s: &str, buf: &mut String) {
    buf.clear();
    for c in s.chars() {
        for lc in c.to_lowercase() {
            buf.push(lc);
        }
    }
}

/// load the first BUFFER*8 (1024k) bytes of the file and check if it is utf8
pub fn isutf8_file(path: &Path) -> Result<bool, CliError> {
    let metadata = std::fs::metadata(path)?;
    let buffer_len = config::DEFAULT_RDR_BUFFER_CAPACITY * 8;
    let file_size = metadata.len() as usize;
    let bytes_to_read: usize = if file_size < buffer_len {
        file_size
    } else {
        buffer_len
    };

    let file = std::fs::File::open(path)?;
    let mut reader = BufReader::new(file);
    let mut buffer = Vec::with_capacity(bytes_to_read);
    reader.read_to_end(&mut buffer)?;

    Ok(simdutf8::basic::from_utf8(&buffer).is_ok())
}

// check if a file is supported by process_input
fn is_supported_file(path: &Path) -> bool {
    // If QSV_SKIP_FORMAT_CHECK is set, consider all files as supported
    if get_envvar_flag("QSV_SKIP_FORMAT_CHECK") {
        return true;
    }

    let ext = path
        .extension()
        .and_then(std::ffi::OsStr::to_str)
        .map(str::to_lowercase)
        .unwrap_or_default();
    match ext.as_str() {
        "csv" | "ssv" | "tsv" | "tab" => true,
        _ => get_special_format(path) != SpecialFormat::Unknown,
    }
}

/// Process the input files and return a vector of paths to the input files.
///
/// If the input is empty, try to copy stdin to a file named stdin in the passed temp directory.
/// If the input is empty and stdin is empty, return an error.
/// If it's not empty, check the input files if they exist, and return an error if they don't.
///
/// If the input is a directory, add all the files in the directory to the input.
/// If the input is a zip file, add all the files in the zip file to the input.
/// If the input is a file with the extension ".infile-list", read the file & add each line as a
/// file to the input.
/// If the input is a file, add the file to the input.
/// If the input are snappy compressed files, uncompress them before adding them to the input.
pub fn process_input(
    arg_input: Vec<PathBuf>,
    tmpdir: &tempfile::TempDir,
    custom_empty_stdin_errmsg: &str,
) -> Result<Vec<PathBuf>, CliError> {
    let mut processed_input = Vec::with_capacity(arg_input.len());

    let work_input = if arg_input.len() == 1 {
        let input_path = &arg_input[0];
        if input_path.is_dir() {
            // if the input is a directory, add all the supported files in the directory to the
            // input
            std::fs::read_dir(input_path)?
                .map(|entry| entry.map(|e| e.path()))
                .filter_map(|path| path.ok().filter(|p| is_supported_file(p)))
                .collect::<Vec<_>>()
        } else if input_path.is_file() {
            // if the input is a file and has the extension "infile-list" case-insensitive,
            // read the file. Each line is a file path
            if input_path
                .extension()
                .is_some_and(|ext| ext.eq_ignore_ascii_case("infile-list"))
            {
                let mut input_file = std::fs::File::open(input_path)?;
                let mut input_file_contents = String::new();
                let mut canonical_invalid_path = PathBuf::new();
                let mut invalid_files = 0_u32;
                input_file.read_to_string(&mut input_file_contents)?;
                let infile_list_vec = input_file_contents
                    .lines()
                    .filter(|line| !line.trim().is_empty() && !line.starts_with('#'))
                    .map(PathBuf::from)
                    .filter_map(|path| {
                        if path.exists() {
                            Some(path)
                        } else {
                            // note that we're warn logging if files do not exist for
                            // each line in the infile-list file
                            // even though we're returning an error on the FIRST file that
                            // doesn't exist in the next section. This is because
                            // we want to log ALL the invalid file paths in the infile-list
                            // file, not just the first one.
                            invalid_files += 1;
                            canonical_invalid_path = path.canonicalize().unwrap_or_default();
                            log::warn!(
                                ".infile-list file '{}': '{}' does not exist",
                                path.display(),
                                canonical_invalid_path.display()
                            );
                            None
                        }
                    })
                    .collect::<Vec<_>>();
                log::info!(
                    ".infile-list file parsed. Filecount - valid:{} invalid:{invalid_files}",
                    infile_list_vec.len()
                );
                infile_list_vec
            } else {
                // if the input is not an ".infile-list" file, add the file to the input
                arg_input
            }
        } else {
            arg_input
        }
    } else {
        arg_input
    };

    let mut stdin_path = PathBuf::new();
    let mut stdin_file_created = false;

    // check the input files
    for path in work_input {
        // check if the path is "-" (stdin)
        if &path == "-" {
            if !stdin_file_created {
                // if stdin was not copied to a file, copy stdin to a file named "stdin"
                let tmp_filename = tmpdir.path().join("stdin.csv");
                let mut tmp_file = std::fs::File::create(&tmp_filename)?;
                std::io::copy(&mut std::io::stdin(), &mut tmp_file)?;
                tmp_file.flush()?;
                stdin_file_created = true;
                stdin_path = tmp_filename;
            }
            processed_input.push(stdin_path.clone());
            continue;
        }

        // Resolve a `dc:<name>` disk-cache reference (the `get` command's cache)
        // to its materialized CSV path BEFORE the existence check below. Without
        // this, the literal "dc:…" string fails `path.exists()`, so every command
        // that routes inputs through `process_input` (cat, slice, join, …) wrongly
        // rejected a valid `dc:` handle even though `Config::new` resolves it. The
        // resolved temp CSV ships a sibling .idx, so indexed commands work too.
        #[cfg(feature = "get")]
        if let Some(dc_name) = path.to_str().and_then(|s| s.strip_prefix("dc:")) {
            processed_input.push(crate::diskcache::resolve_dc_path(dc_name)?);
            continue;
        }

        if !path.exists() {
            return fail_clierror!("Input file '{}' does not exist", path.display());
        }

        // is the input file snappy compressed?
        if path
            .extension()
            .is_some_and(|ext| ext.eq_ignore_ascii_case("sz"))
        {
            // if so, decompress the file
            let decompressed_filepath = decompress_snappy_file(&path, tmpdir)?;

            // rename the decompressed file to the original filename, but still
            // inside the temp directory. this is so that the decompressed file can be
            // processed as if it was the original file without the "sz" extension
            let original_filepath = path.with_extension("");
            // safety: we know the path has a filename
            let original_filename = original_filepath.file_name().unwrap();

            let final_decompressed_filepath = tmpdir.path().join(original_filename);
            std::fs::rename(&decompressed_filepath, &final_decompressed_filepath)?;

            processed_input.push(final_decompressed_filepath);
        }
        // is the input file a zip archive?
        else if path
            .extension()
            .is_some_and(|ext| ext.eq_ignore_ascii_case("zip"))
        {
            // Extract its usable entries (tabular-first) via the shared zip module
            // so this command-level "all entries" path and the reader-level
            // `Config`/`select_zip_entry` "first tabular entry" path agree on what
            // a zip's entries are. Issue #3988.
            log::info!("Extracting files from zip archive: {}", path.display());
            let entries = extract_all_zip_entries(&path, tmpdir)?;
            log::info!(
                "Extracted {} usable entries from zip archive {}",
                entries.len(),
                path.display()
            );
            processed_input.extend(entries);
        } else {
            processed_input.push(path);
        }
    }

    if processed_input.is_empty() {
        if custom_empty_stdin_errmsg.is_empty() {
            return fail_clierror!(
                "No data on stdin. Please provide at least one input file or pipe data to stdin."
            );
        }
        return fail_clierror!("{custom_empty_stdin_errmsg}");
    }
    log::debug!("processed input file/s: {processed_input:?}");
    Ok(processed_input)
}

#[inline]
pub fn replace_column_value(
    record: &csv::StringRecord,
    column_index: usize,
    new_value: &str,
) -> csv::StringRecord {
    record
        .into_iter()
        .enumerate()
        .map(|(i, v)| if i == column_index { new_value } else { v })
        .collect()
}

/// format a `SystemTime` from a file's metadata to a string using the format specifier
#[inline]
pub fn format_systemtime(time: SystemTime, format_specifier: &str) -> String {
    // Default to UNIX_EPOCH for pre-1970 timestamps. They do occur in the wild —
    // Windows files can carry the FILETIME epoch (1601-01-01), and archived /
    // restored files or corrupted FS metadata sometimes report pre-1970 mtimes.
    let timestamp = time
        .duration_since(SystemTime::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs();

    let datetime = chrono::DateTime::from_timestamp(timestamp as i64, 0).unwrap_or_default();
    format!("{datetime}", datetime = datetime.format(format_specifier))
}

pub fn create_json_writer(
    output: Option<&String>,
    buffer_capacity: usize,
) -> std::io::Result<Box<dyn Write + Send + 'static>> {
    // create a JSON writer
    // if flag_output is None or "-" then write to stdout
    let output = output.as_ref().map_or("-", |s| s.as_str());
    let buffer_size = if buffer_capacity == 0 {
        config::DEFAULT_WTR_BUFFER_CAPACITY
    } else {
        buffer_capacity
    };
    let writer: Box<dyn Write + Send + 'static> = match output {
        "-" => Box::new(std::io::BufWriter::with_capacity(
            buffer_size,
            std::io::stdout(),
        )),
        "stderr" => Box::new(std::io::BufWriter::with_capacity(
            buffer_size,
            std::io::stderr(),
        )),
        _ => Box::new(std::io::BufWriter::with_capacity(
            buffer_size,
            fs::File::create(output)?,
        )),
    };
    Ok(writer)
}

/// iterate over the CSV `ByteRecords` and write them to the JSON file.
///
/// Records are taken as `csv::Result<ByteRecord>` so callers can pass the
/// raw `byte_records()` iterator and have parse errors propagate with `?`
/// without buffering the entire slice in memory first.
///
/// **Failure semantics:** because records stream straight to the writer,
/// a CSV parse error mid-stream means the opening `[` and any prior records
/// will already have been written before the error propagates. The closing
/// `]` is only written after the loop completes cleanly, so the output
/// (stdout or `output` file) may be a truncated, syntactically invalid
/// JSON document on error. Callers that need all-or-nothing semantics must
/// validate / collect records before invoking this.
pub fn write_json(
    output: Option<&String>,
    no_headers: bool,
    headers: &csv::ByteRecord,
    records: impl Iterator<Item = csv::Result<csv::ByteRecord>>,
) -> CliResult<()> {
    let mut json_wtr = create_json_writer(output, config::DEFAULT_WTR_BUFFER_CAPACITY * 4)?;

    // keys are escaped once here, per JSON spec. Note that the escaped form
    // returned by serde_json includes the enclosing double quotes, so the
    // format string below must NOT add its own quotes around the key.
    let header_vec: Vec<String> = headers
        .iter()
        .enumerate()
        .map(|(col_idx, b)| {
            let key = if no_headers {
                col_idx.to_string()
            } else {
                bytes_to_cow_str(b).into_owned()
            };
            serde_json::Value::String(key).to_string()
        })
        .collect();

    // Write the opening bracket for the JSON array
    write!(json_wtr, "[")?;
    let mut is_first = true;

    let mut temp_val;
    let null_val = "null".to_string();
    let mut json_string_val: serde_json::Value;

    for record in records {
        let record = record?;
        if is_first {
            is_first = false;
        } else {
            // Write a comma before each record except the first one
            write!(json_wtr, ",")?;
        }
        write!(json_wtr, "{{")?;
        // Emit field separators using a leading-comma join: write a comma
        // *before* each non-first field. This keeps the output valid for
        // both flexible-CSV cases — records longer than headers (skip
        // extras) and records shorter than headers (no trailing comma).
        let mut first_field = true;
        for (idx, b) in record.iter().enumerate() {
            let Some(key) = header_vec.get(idx) else {
                break;
            };
            temp_val = bytes_to_cow_str(b).into_owned();
            if temp_val.is_empty() {
                temp_val.clone_from(&null_val);
            } else {
                // we round-trip the value to serde_json
                // to escape the string properly per JSON spec
                json_string_val = serde_json::Value::String(temp_val);
                temp_val = json_string_val.to_string();
            }
            if first_field {
                first_field = false;
            } else {
                write!(&mut json_wtr, ",")?;
            }
            write!(&mut json_wtr, r#"{key}:{temp_val}"#)?;
        }
        write!(json_wtr, "}}")?;
    }
    // Write the closing bracket for the JSON array
    writeln!(json_wtr, "]")?;

    Ok(json_wtr.flush()?)
}

/// write a single `csv::ByteRecord` to a JSON record writer
/// if `no_headers` is true, the column index (0-based) is used as the key
/// if `no_headers` is false, the header is used as the key
/// if `is_first` is true, a comma is not written before the record
/// if `is_first` is false, a comma is written before the record
/// `is_first` is passed as a mutable reference so that it can be updated
/// in this helper function efficiently
/// in this way, we can stream JSON records to a writer
pub fn write_json_record<W: std::io::Write>(
    json_wtr: &mut W,
    no_headers: bool,
    headers: &csv::ByteRecord,
    record: &csv::ByteRecord,
    is_first: &mut bool,
) -> std::io::Result<()> {
    // keys are escaped once here, per JSON spec. Note that the escaped form
    // returned by serde_json includes the enclosing double quotes, so the
    // format strings below must NOT add their own quotes around the key.
    let header_vec: Vec<String> = headers
        .iter()
        .enumerate()
        .map(|(col_idx, b)| {
            let key = if no_headers {
                col_idx.to_string()
            } else {
                bytes_to_cow_str(b).into_owned()
            };
            serde_json::Value::String(key).to_string()
        })
        .collect();

    let rec_len = header_vec.len().saturating_sub(1);
    let mut temp_val;
    let mut json_string_val: serde_json::Value;
    let null_val = "null".to_string();

    if *is_first {
        write!(json_wtr, "{{")?;
        *is_first = false;
    } else {
        write!(json_wtr, ",{{")?;
    }
    for (idx, b) in record.iter().enumerate() {
        temp_val = bytes_to_cow_str(b).into_owned();
        if temp_val.is_empty() {
            temp_val.clone_from(&null_val);
        } else {
            json_string_val = serde_json::Value::String(temp_val);
            temp_val = json_string_val.to_string();
        }
        if idx < rec_len {
            unsafe {
                write!(
                    json_wtr,
                    r#"{key}:{value},"#,
                    key = header_vec.get_unchecked(idx),
                    value = temp_val
                )?;
            }
        } else {
            unsafe {
                write!(
                    json_wtr,
                    r#"{key}:{value}"#,
                    key = header_vec.get_unchecked(idx),
                    value = temp_val
                )?;
            }
        }
    }
    Ok(write!(json_wtr, "}}")?)
}

/// get stats records from stats.csv.data.jsonl file, or if its invalid, by running the stats
/// command returns tuple (`csv_fields`, `csv_stats`)
/// Does a reused stats cache contain everything `mode` needs?
///
/// The stats cache (`<input>.stats.csv.data.jsonl`) is keyed only by
/// mtime, not by the `StatsMode` that produced it, so a cache built by a
/// leaner mode can be reused by a richer one. Most modes only need the
/// base cardinality/types set any cache carries, so they always pass.
/// `StatsMode::ProfileSchema` additionally needs mode + quartiles; a cache
/// from `qsv schema` (`StatsMode::Schema`) lacks both. Since no
/// producing-mode marker is stored, sufficiency is detected from the data:
///   * `--mode` populates the `mode_count` metadata field for every column (even `Some(0)` for an
///     all-unique column, where the `mode` value itself is `None`); a lean cache leaves
///     `mode_count` `None`. We test `mode_count`, not `mode`, so all-unique datasets aren't
///     mistaken for a lean cache and needlessly regenerated on every run.
///   * `--quartiles` populates `q2_median` for numeric columns, so require it only when the dataset
///     actually has a numeric column.
///   * `--zero-padded-numeric` is NOT detectable from the data (the column is empty for every
///     non-flagged column, so a dataset with no zero-padded codes looks identical with or without
///     the flag) and is therefore not required here: a pre-existing cache without it is still
///     reused, and `viz smart` simply falls back to its statistical routing for such columns until
///     the cache is naturally regenerated.
fn stats_satisfy_mode(stats: &[StatsData], mode: StatsMode) -> bool {
    if mode != StatsMode::ProfileSchema {
        return true;
    }
    let has_mode = stats.iter().any(|s| s.mode_count.is_some());
    let has_numeric = stats
        .iter()
        .any(|s| s.r#type == "Integer" || s.r#type == "Float");
    let has_quartiles = stats.iter().any(|s| s.q2_median.is_some());
    has_mode && (!has_numeric || has_quartiles)
}

/// Do the parsing options recorded in a stats cache's `<input>.stats.csv.json` metadata match the
/// options the consuming command is using?
///
/// The cache is located by input path and validated by mtime, NOT by parsing options — so a cache
/// built under a different `--no-headers` / `--delimiter` can pass every freshness check while
/// describing a DIFFERENT logical CSV: field names become positional (`0`, `1`, …), the header row
/// is counted as data, and the column count can change entirely. Reusing it silently yields wrong
/// stats, or a hard failure downstream when the column count disagrees.
///
/// Shared by `get_stats_records` and `get_stats_records_readonly` deliberately: those two drifted
/// (only the latter validated this), which is precisely how the mismatch became reachable.
///
/// Returns `false` when the metadata is missing or unreadable, so callers regenerate rather than
/// trust a cache they cannot verify.
fn stats_cache_parsing_opts_match(
    metadata: &serde_json::Value,
    input_path: &Path,
    no_headers: bool,
    delimiter: Option<Delimiter>,
) -> bool {
    // --no-headers determines whether the header row is part of the data (and therefore whether
    // field names are real or positional), so it must match the consuming command.
    if metadata
        .get("flag_no_headers")
        .and_then(serde_json::Value::as_bool)
        .unwrap_or(false)
        != no_headers
    {
        return false;
    }

    // The effective delimiter must match. Both the cache and the command resolve an unset
    // delimiter from the same file path, so compare resolved bytes: an explicit recorded/passed
    // delimiter is a single ASCII byte, otherwise the extension default.
    let ext_delim = {
        let (_, d, _) = get_delim_by_extension(input_path, b',');
        d
    };
    let cache_delim = match metadata.get("flag_delimiter").and_then(|v| v.as_str()) {
        Some(s) if !s.is_empty() => s.as_bytes()[0],
        _ => ext_delim,
    };
    let cmd_delim = delimiter.map_or(ext_delim, |d| d.as_byte());
    cache_delim == cmd_delim
}

/// Does a stats cache's metadata sidecar record parsing options that CONFLICT with the consuming
/// command's? Only a positive, readable mismatch counts as a conflict.
///
/// Absent or unparseable metadata is deliberately NOT a conflict. `moarstats` writes the
/// `.stats.csv.data.jsonl` cache without a companion `.stats.csv.json`, and treating those as
/// unusable would silently discard a rich externally-built cache and regenerate a leaner one —
/// downgrading consumers (e.g. `pivotp --agg smart` loses the `sort_order` it decides on). The
/// mismatch this guards against always writes a sidecar, since it goes through the `stats`
/// subprocess, so declining to judge a sidecar-less cache leaves the pre-existing status quo
/// rather than a new hole.
///
/// (`get_stats_records_readonly` still fails closed on missing metadata: it returns an `Option`
/// its callers treat as a best-effort accelerator, so refusing costs nothing there.)
fn stats_cache_parsing_opts_conflict(
    input_path: &Path,
    canonical_input_path: &Path,
    no_headers: bool,
    delimiter: Option<Delimiter>,
) -> bool {
    // Both locations must be checked, because for a SYMLINKED input they differ: the JSONL cache
    // is looked up beside the canonicalized target, but the `stats` subprocess is invoked with the
    // original path and writes its `<input>.stats.csv.json` beside the symlink. Looking only
    // beside the canonical target finds no metadata there and waves the cache through — which
    // reopened the exact mismatch this guard exists to catch, via a symlink.
    //
    // Delimiter resolution uses `input_path`, the path actually parsed: `Config` resolves an unset
    // delimiter from that path's extension, so a `link.csv -> target.tsv` symlink must be judged
    // by the extension the reader used, not the target's.
    for meta_base in [input_path, canonical_input_path] {
        let metadata_path = meta_base.with_extension("stats.csv.json");
        let Ok(text) = std::fs::read_to_string(&metadata_path) else {
            continue;
        };
        let Ok(metadata) = serde_json::from_str::<serde_json::Value>(&text) else {
            continue;
        };
        if !stats_cache_parsing_opts_match(&metadata, input_path, no_headers, delimiter) {
            return true;
        }
    }
    false
}

pub fn get_stats_records(
    args: &SchemaArgs,
    requested_mode: StatsMode,
) -> CliResult<(ByteRecord, Vec<StatsData>)> {
    let env_mode = env::var("QSV_STATSCACHE_MODE")
        .unwrap_or_else(|_| DEFAULT_STATSCACHE_MODE.to_string())
        .to_ascii_lowercase();

    if !["auto", "force", "none"].contains(&env_mode.as_str()) {
        return fail_incorrectusage_clierror!(
            "Invalid QSV_STATSCACHE_MODE value: {env_mode}. Must be one of: auto, force, none"
        );
    }

    if requested_mode == StatsMode::None
        || env_mode == "none"
        || args.arg_input.is_none()
        || args.arg_input.as_ref() == Some(&"-".to_string())
        // safety: we know that by this point, args.arg_input is not None as
        // the earlier is_none() check would have short-circuited already
        || get_special_format(Path::new(args.arg_input.as_ref().unwrap())) != SpecialFormat::Unknown
    {
        // if stdin or StatsMode::None,
        // we're just doing frequency old school w/o cardinality
        return Ok((ByteRecord::new(), Vec::new()));
    }

    let input_path_raw = args.arg_input.as_ref().ok_or("No input provided")?;
    // Resolve a `dc:<name>` disk-cache handle (the `get` command's cache) to its
    // materialized CSV path: the stats cache is keyed/located by the concrete
    // file path, so `canonicalize`, the `.stats.csv.data.jsonl` sidecar location,
    // and the stats subprocess input all need a real path, not the "dc:" string.
    #[cfg(feature = "get")]
    let input_path_owned: String = if let Some(dc_name) = input_path_raw.strip_prefix("dc:") {
        crate::diskcache::resolve_dc_path(dc_name)?
            .to_string_lossy()
            .into_owned()
    } else {
        input_path_raw.clone()
    };
    #[cfg(not(feature = "get"))]
    let input_path_owned: String = input_path_raw.clone();
    let input_path = &input_path_owned;
    let canonical_input_path = Path::new(input_path).canonicalize()?;
    let statsdata_path = canonical_input_path.with_extension("stats.csv.data.jsonl");

    let stats_data_current = if statsdata_path.exists() {
        let statsdata_metadata = std::fs::metadata(&statsdata_path)?;

        let input_metadata = std::fs::metadata(input_path)?;

        let statsdata_mtime = FileTime::from_last_modification_time(&statsdata_metadata);
        let input_mtime = FileTime::from_last_modification_time(&input_metadata);
        if statsdata_mtime > input_mtime {
            // mtime alone is not sufficient: the cache is NOT keyed by parsing options, so a
            // cache written by an earlier run with a different --no-headers/--delimiter would
            // otherwise be reused here and describe a different logical CSV.
            if stats_cache_parsing_opts_conflict(
                Path::new(input_path),
                &canonical_input_path,
                args.flag_no_headers,
                args.flag_delimiter,
            ) {
                info!(
                    "stats.csv.data.jsonl file was built with different parsing options \
                     (--no-headers/--delimiter). Regenerating stats jsonl."
                );
                false
            } else {
                info!("Valid stats.csv.data.jsonl file found!");
                true
            }
        } else {
            info!("stats.csv.data.jsonl file is older than input file. Regenerating stats jsonl.");
            false
        }
    } else {
        info!(
            "stats.csv.data.jsonl file does not exist: {}",
            statsdata_path.display()
        );
        false
    };

    if requested_mode == StatsMode::Frequency && env_mode != "auto" && !stats_data_current {
        // if the stats.data file is not current,
        // we're also doing frequency old school w/o cardinality
        // unless env_mode auto overrides
        return Ok((ByteRecord::new(), Vec::new()));
    }

    // Use qsv's Config system to properly detect delimiter based on file extension
    let rconfig = Config::new(Some(input_path))
        .delimiter(args.flag_delimiter)
        .no_headers_flag(args.flag_no_headers);
    let mut rdr = rconfig.reader()?;
    // get the headers from the input file
    let csv_fields = rdr.byte_headers()?.clone();
    drop(rdr);

    // Update args with the detected delimiter if it wasn't explicitly set
    let detected_delimiter = if args.flag_delimiter.is_none() {
        let path = Path::new(input_path);
        let (_, detected_delim, _) = get_delim_by_extension(path, b',');
        Some(Delimiter(detected_delim))
    } else {
        args.flag_delimiter
    };

    let mut stats_data_loaded = false;
    let mut csv_stats: Vec<StatsData> = Vec::with_capacity(csv_fields.len());

    // if stats_data file exists and is current, use it
    if stats_data_current && !args.flag_force {
        let statsdatajson_rdr =
            BufReader::with_capacity(DEFAULT_RDR_BUFFER_CAPACITY, File::open(statsdata_path)?);

        let mut curr_line: String;
        let mut s_slice: Vec<u8>;

        for line in statsdatajson_rdr.lines() {
            curr_line = line?;
            s_slice = curr_line.as_bytes().to_vec();

            // Parse regular stats record
            let parse_result =
                cfg_select! {
                    target_endian = "little" => simd_json::from_slice::<StatsData>(&mut s_slice),
                    _ => serde_json::from_slice::<StatsData>(&s_slice),
                };

            if let Ok(stats) = parse_result {
                csv_stats.push(stats);
            } else {
                // if we encounter a parsing error, clear csv_stats and break
                // so that we regenerate the stats data
                csv_stats.clear();
                break;
            }
        }
        stats_data_loaded = !csv_stats.is_empty();

        // Mode-aware cache reuse: the cache is keyed only by mtime, not by
        // the StatsMode that produced it. A cache built by a leaner mode
        // (e.g. `qsv schema`, StatsMode::Schema — no --quartiles/--mode)
        // doesn't satisfy ProfileSchema's need for quartiles + mode. If the
        // loaded stats lack what the requested mode requires, discard them
        // so the regeneration path below recomputes with the right flags.
        if stats_data_loaded && !stats_satisfy_mode(&csv_stats, requested_mode) {
            csv_stats.clear();
            stats_data_loaded = false;
        }
    }

    // otherwise, run stats command to generate stats.csv.data.jsonl file
    if !stats_data_loaded {
        let stats_args = crate::cmd::stats::Args {
            // the dc:-resolved concrete path, so the stats subprocess reads the
            // real (materialized) file rather than re-resolving the "dc:" handle
            arg_input:                 Some(input_path.clone()),
            flag_select:               crate::select::SelectColumns::parse("").unwrap(),
            flag_everything:           false,
            flag_typesonly:            false,
            flag_infer_boolean:        false,
            flag_boolean_patterns:     String::new(),
            flag_mode:                 false,
            flag_cardinality:          true,
            flag_zero_padded_numeric:  false,
            flag_median:               false,
            flag_quartiles:            false,
            flag_mad:                  false,
            flag_percentiles:          false,
            flag_percentile_list:      "5,10,40,60,90,95".to_string(),
            flag_quantile_method:      "exact".to_string(),
            flag_cardinality_method:   "exact".to_string(),
            flag_mode_cardinality_cap: 0,
            flag_nulls:                false,
            flag_round:                4,
            flag_infer_dates:          true,
            flag_dates_whitelist:      args.flag_dates_whitelist.to_string(),
            flag_prefer_dmy:           args.flag_prefer_dmy,
            flag_force:                args.flag_force,
            flag_jobs:                 Some(njobs(args.flag_jobs)),
            flag_stats_jsonl:          true,
            flag_jsonl:                false,
            flag_pretty_json:          false,
            flag_cache_threshold:      1, // force the creation of stats cache files
            flag_output:               None,
            flag_no_headers:           args.flag_no_headers,
            flag_delimiter:            detected_delimiter,
            flag_memcheck:             args.flag_memcheck,
            flag_vis_whitespace:       false,
            flag_weight:               None,
        };

        let tempfile = tempfile::Builder::new().suffix(".stats.csv").tempfile()?;
        // safety: we just created a tempfile, which is guaranteed to have a path
        let tempfile_path = tempfile.path().to_str().unwrap().to_string();

        let statsdatajson_path = &canonical_input_path.with_extension("stats.csv.data.jsonl");

        let input = stats_args.arg_input.unwrap_or_else(|| "-".to_string());

        // for the date-inferring modes (Schema/ProfileSchema), default an unspecified
        // whitelist to "sniff" so date/datetime columns are auto-detected rather than left as
        // strings. An explicit caller value (e.g. `qsv schema --dates-whitelist`, profile's
        // preset, viz's "sniff") still wins.
        let dates_whitelist = if stats_args.flag_dates_whitelist.is_empty() {
            "sniff"
        } else {
            stats_args.flag_dates_whitelist.as_str()
        };

        // we do rustfmt::skip here as it was breaking the stats cmdline along strange
        // boundaries, causing CI errors.
        // This is because we're using tab characters (/t) to separate args to fix #2294,
        #[rustfmt::skip]
        let mut stats_args_str = match requested_mode {
            StatsMode::Schema => {
                // mode is StatsMode::Schema
                // we're generating schema, so we need cardinality and to infer-dates
                format!(
                    "stats\t{input}\t--round\t4\t--cardinality\
                    \t--infer-dates\t--dates-whitelist\t{dates_whitelist}\
                    \t--stats-jsonl\t--force\t--output\t{tempfile_path}"
                )
            },
            StatsMode::ProfileSchema => {
                // mode is StatsMode::ProfileSchema
                // same as Schema, plus quartiles (q1/median/q3) and mode so
                // `qsv profile`'s descriptive-statistics projection has the
                // full extended stat set without a pre-built stats cache, and
                // zero-padded-numeric detection (a streaming, constant-memory
                // stat) so `viz smart` can route zip/FIPS/ICD-style code
                // columns as dimensions instead of measures.
                format!(
                    "stats\t{input}\t--round\t4\t--cardinality\
                    \t--quartiles\t--mode\t--zero-padded-numeric\
                    \t--infer-dates\t--dates-whitelist\t{dates_whitelist}\
                    \t--stats-jsonl\t--force\t--output\t{tempfile_path}"
                )
            },
            StatsMode::Frequency => {
                // StatsMode::Frequency
                // we're doing frequency, so we just need cardinality
                format!("stats\t{input}\t--cardinality\t--stats-jsonl\t--output\t{tempfile_path}")
            },
            StatsMode::FrequencyForceStats => {
                // StatsMode::FrequencyForceStats
                // we're doing frequency, so we need cardinality from a --forced stats run
                format!(
                    "stats\t{input}\t--cardinality\t--stats-jsonl\t--force\t--output\t{tempfile_path}"
                )
            },
            #[cfg(feature = "polars")]
            StatsMode::PolarsSchema => {
                // StatsMode::PolarsSchema
                // we need data types, ranges & cardinality
                // if sniff detected date columns, also infer dates with the sniffed whitelist
                if stats_args.flag_dates_whitelist.is_empty() {
                    format!("stats\t{input}\t--cardinality\t--stats-jsonl\t--output\t{tempfile_path}")
                } else {
                    format!(
                        "stats\t{input}\t--cardinality\
                        \t--infer-dates\t--dates-whitelist\t{dates_whitelist}\
                        \t--stats-jsonl\t--output\t{tempfile_path}",
                        dates_whitelist = stats_args.flag_dates_whitelist
                    )
                }
            },
            StatsMode::Outliers => {
                // StatsMode::Outliers
                // we need data types, ranges, cardinality, quartiles, mad and modes/antimodes
                format!("stats\t{input}\t--cardinality\t--quartiles\t--mad\t--mode\t--stats-jsonl\t--output\t{tempfile_path}")
            },
            StatsMode::None => unreachable!(), // we returned early on None earlier
        };
        if args.flag_prefer_dmy {
            stats_args_str = format!("{stats_args_str}\t--prefer-dmy");
        }
        if args.flag_no_headers {
            stats_args_str = format!("{stats_args_str}\t--no-headers");
        }

        // Use the detected delimiter
        stats_args_str = format!("{stats_args_str}\t--delimiter\t{}", {
            // safety: we know it's Some because we set it above
            let delim_to_use = detected_delimiter.unwrap().as_byte();
            if delim_to_use == b'\t' {
                r"\t".to_string()
            } else {
                (delim_to_use as char).to_string()
            }
        });

        if args.flag_memcheck {
            stats_args_str = format!("{stats_args_str}\t--memcheck");
        }
        if let Some(jobs) = stats_args.flag_jobs {
            stats_args_str = format!("{stats_args_str}\t--jobs\t{jobs}");
        }
        if stats_args.flag_nulls {
            stats_args_str = format!("{stats_args_str}\t--nulls");
        }

        if env_mode == "force" && !stats_args_str.contains("--force") {
            stats_args_str = format!("{stats_args_str}\t--force");
        }

        let stats_args_vec: Vec<&str> = stats_args_str.split('\t').collect();

        let qsv_bin = current_exe()?;
        let mut stats_cmd = std::process::Command::new(qsv_bin);
        if requested_mode == StatsMode::Outliers {
            // set the max length for antimodes
            stats_cmd.env("QSV_ANTIMODES_LEN", "0").args(stats_args_vec);
        } else {
            stats_cmd.args(stats_args_vec);
        }
        let status = stats_cmd.output()?.status;
        if !status.success() {
            let status_code = status.code();
            if let Some(code) = status_code {
                return Err(CliError::Other(format!(
                    "qsv stats exited with code: {code}"
                )));
            }
            return Err(CliError::Other(cfg_select! {
                target_family = "unix" => {
                    match status.signal() {
                        Some(signal) => format!("qsv stats terminated with signal: {signal}"),
                        None => "qsv stats terminated by unknown cause".to_string(),
                    }
                },
                _ => {
                    "qsv stats terminated by unknown cause".to_string()
                },
            }));
        }

        // create a stats data jsonl from the output of the stats command
        // stats command always writes comma-delimited CSV to tempfile,
        // so b',' is correct here regardless of the original input delimiter
        csv_to_jsonl(
            &tempfile_path,
            &STATSDATA_TYPES_MAP,
            statsdatajson_path,
            b',',
        )?;

        // Keep the metadata sidecar beside the JSONL we just wrote in sync with it.
        //
        // We write the JSONL at the CANONICAL path, but the `stats` subprocess writes its
        // `<input>.stats.csv.json` args sidecar beside the path it was GIVEN. Those coincide for
        // an ordinary input and diverge for a symlink — and a divergence is not merely a missing
        // sidecar: an earlier direct run on the target leaves a canonical sidecar that keeps
        // describing itself as current while this run replaces the canonical JSONL underneath it.
        // A later direct run then reads stale-but-matching metadata and reuses a cache built with
        // different parsing options. Copy the sidecar that actually describes this JSONL over it.
        let subprocess_metadata = Path::new(input_path).with_extension("stats.csv.json");
        let canonical_metadata = canonical_input_path.with_extension("stats.csv.json");
        // Compare RESOLVED paths, not the raw ones: for an ordinary input these name the same
        // file via a relative and an absolute path, and `fs::copy` onto itself truncates it —
        // which corrupts the very metadata this is meant to keep trustworthy.
        let same_file = match (
            subprocess_metadata.canonicalize(),
            canonical_metadata.canonicalize(),
        ) {
            (Ok(a), Ok(b)) => a == b,
            _ => false,
        };
        if !same_file && subprocess_metadata.exists() {
            // best-effort: a stale sidecar is only a cache-reuse hint, never required for
            // correctness of THIS run, so a copy failure must not fail the stats load
            if let Err(e) = std::fs::copy(&subprocess_metadata, &canonical_metadata) {
                log::warn!(
                    "could not sync stats cache metadata to {}: {e}",
                    canonical_metadata.display()
                );
            }
        }

        let statsdatajson_rdr =
            BufReader::with_capacity(DEFAULT_RDR_BUFFER_CAPACITY, File::open(statsdatajson_path)?);

        let mut curr_line: String;
        let mut s_slice: Vec<u8>;
        for line in statsdatajson_rdr.lines() {
            curr_line = line?;
            s_slice = curr_line.as_bytes().to_vec();

            // Parse regular stats record
            let parse_result =
                cfg_select! {
                    target_endian = "little" => simd_json::from_slice::<StatsData>(&mut s_slice),
                    _ => serde_json::from_slice::<StatsData>(&s_slice),
                };

            match parse_result {
                Ok(stats) => csv_stats.push(stats),
                Err(e) => return Err(CliError::Other(format!("error parsing stats: {e}"))),
            }
        }
    }

    if csv_stats.len() != csv_fields.len() {
        // stats cache is likely corrupted or truncated; fall back to empty stats
        return Ok((ByteRecord::new(), Vec::new()));
    }
    Ok((csv_fields.iter().take(csv_stats.len()).collect(), csv_stats))
}

/// Opportunistically load an EXISTING, CURRENT stats cache sidecar
/// (`<input>.stats.csv.data.jsonl`) WITHOUT ever spawning a `qsv stats`
/// subprocess. Used by stats-cache-aware commands (e.g. `extsort`, `sortcheck`)
/// to short-circuit work when the cache already proves a column's sort order.
///
/// Returns `None` (never an error) when the cache is absent, stale, disabled via
/// `QSV_STATSCACHE_MODE=none`, the input is stdin / a special format, or any
/// IO/parse problem occurs. Unlike [`get_stats_records`], it NEVER regenerates
/// the cache.
#[allow(clippy::missing_panics_doc)]
pub fn get_stats_records_readonly(
    input_path: Option<&str>,
    no_headers: bool,
    delimiter: Option<Delimiter>,
) -> Option<Vec<StatsData>> {
    // honor the stats-cache opt-out (same mechanism as frequency/schema/profile)
    let env_mode = env::var("QSV_STATSCACHE_MODE")
        .unwrap_or_else(|_| DEFAULT_STATSCACHE_MODE.to_string())
        .to_ascii_lowercase();
    if env_mode == "none" {
        return None;
    }

    let input_path_raw = input_path?;
    // reject stdin and special formats (snappy/zip/parquet/etc.) - the stats
    // cache is keyed/located by a concrete CSV/TSV path
    if input_path_raw == "-"
        || get_special_format(Path::new(input_path_raw)) != SpecialFormat::Unknown
    {
        return None;
    }

    // resolve a `dc:<name>` disk-cache handle (the `get` command's cache) to its
    // materialized CSV path, mirroring get_stats_records
    #[cfg(feature = "get")]
    let input_path_owned: String = if let Some(dc_name) = input_path_raw.strip_prefix("dc:") {
        crate::diskcache::resolve_dc_path(dc_name)
            .ok()?
            .to_string_lossy()
            .into_owned()
    } else {
        input_path_raw.to_string()
    };
    #[cfg(not(feature = "get"))]
    let input_path_owned: String = input_path_raw.to_string();

    let canonical_input_path = Path::new(&input_path_owned).canonicalize().ok()?;
    let statsdata_path = canonical_input_path.with_extension("stats.csv.data.jsonl");

    // the cache sidecar must exist AND be newer than the input file
    let statsdata_metadata = std::fs::metadata(&statsdata_path).ok()?;
    let input_metadata = std::fs::metadata(&input_path_owned).ok()?;
    let statsdata_mtime = FileTime::from_last_modification_time(&statsdata_metadata);
    let input_mtime = FileTime::from_last_modification_time(&input_metadata);
    if statsdata_mtime <= input_mtime {
        return None;
    }

    // Validate the companion args metadata (<input>.stats.csv.json). The JSONL
    // sidecar's sort_order / column-mapping is only valid for the parser options it
    // was generated with: a cache built with a different --no-headers, delimiter, or
    // a column --select can pass the mtime/count checks yet describe a different
    // logical CSV. Fail closed if the metadata is missing, unreadable, or mismatched.
    let metadata_path = canonical_input_path.with_extension("stats.csv.json");
    let metadata: serde_json::Value =
        serde_json::from_str(&std::fs::read_to_string(&metadata_path).ok()?).ok()?;

    // only a full-file stats cache can be trusted for positional column mapping
    // (a `--select`ed cache may drop or reorder columns relative to the header)
    if metadata.get("flag_select").and_then(|v| v.as_str()) != Some("<All>") {
        return None;
    }
    // --no-headers and the effective delimiter must match the consuming command; shared with
    // `get_stats_records` so the two cannot drift apart again.
    if !stats_cache_parsing_opts_match(
        &metadata,
        Path::new(&input_path_owned),
        no_headers,
        delimiter,
    ) {
        return None;
    }

    // get the input header count so we can detect a truncated/corrupt cache
    let rconfig = Config::new(Some(&input_path_owned))
        .delimiter(delimiter)
        .no_headers_flag(no_headers);
    let header_count = rconfig.reader().ok()?.byte_headers().ok()?.len();

    // parse the jsonl sidecar - one StatsData record per line, in header order
    let statsdatajson_rdr = BufReader::with_capacity(
        DEFAULT_RDR_BUFFER_CAPACITY,
        File::open(&statsdata_path).ok()?,
    );
    let mut csv_stats: Vec<StatsData> = Vec::with_capacity(header_count);
    for line in statsdatajson_rdr.lines() {
        let curr_line = line.ok()?;
        if curr_line.trim().is_empty() {
            continue;
        }
        // Parse leniently via serde_json::Value rather than strictly into StatsData:
        // an opportunistic cache may have been produced by a lean `stats --stats-jsonl`
        // run that omits non-Option fields like `cardinality`, which would make a strict
        // StatsData deserialize fail. We only need a handful of fields to decide a sort
        // short-circuit, so we extract just those (defaulting the rest).
        let v: serde_json::Value = serde_json::from_slice(curr_line.as_bytes()).ok()?;
        let stats = StatsData {
            field: v
                .get("field")
                .and_then(|x| x.as_str())
                .unwrap_or_default()
                .to_string(),
            r#type: v.get("type").and_then(|x| x.as_str())?.to_string(),
            is_ascii: v
                .get("is_ascii")
                .and_then(serde_json::Value::as_bool)
                .unwrap_or(false),
            sort_order: v
                .get("sort_order")
                .and_then(|x| x.as_str())
                .map(ToString::to_string),
            // nullcount is a base streaming stat always present; fail closed if absent
            nullcount: v.get("nullcount").and_then(serde_json::Value::as_u64)?,
            ..Default::default()
        };
        csv_stats.push(stats);
    }

    if csv_stats.is_empty() || csv_stats.len() != header_count {
        return None;
    }
    Some(csv_stats)
}

/// Does the stats cache *prove* the single selected column is already in
/// **ascending** order under the requested comparator semantics? Returns `false`
/// when not provably so (the caller must then fall back to the full operation,
/// which is always correct).
///
/// Only the ascending case is decided here. Descending (`--reverse`) is
/// deliberately not short-circuited: `extsort --reverse` reverses the full
/// `sort_key|position` key, so equal selected keys are emitted in reverse input
/// order (an anti-stable tie-break) that a straight passthrough cannot reproduce.
/// Ascending, by contrast, breaks ties by ascending position == input order,
/// which passthrough preserves exactly.
///
/// The cache's `sort_order` is computed using each column's inferred type
/// (String -> byte-lexicographic; Integer/Float -> numeric; Date/DateTime ->
/// chronological). A short-circuit is only safe when the caller's comparator
/// semantics exactly match the cache's per-column semantics, so ALL of the
/// following must hold:
/// - exactly one selected column (per-column order can't prove multi-column order)
/// - `nullcount == 0` (empty cells are excluded from `sort_order` but sort as Less)
/// - `sort_order` is present (all-null columns have none)
/// - type matches the requested comparator (and ASCII when `for_extsort`)
/// - the cached order is `Ascending`
///
/// `requested_numeric`: caller wants numeric comparison (vs. byte-lexicographic).
/// `for_extsort`: additionally require the column to be ASCII - extsort CSV mode
///   concatenates fields via lossy UTF-8, so only ASCII guarantees byte order
///   equals String order.
#[must_use]
pub fn stats_cache_proves_ascending(
    sel: &[usize],
    stats: &[StatsData],
    requested_numeric: bool,
    for_extsort: bool,
) -> bool {
    // a per-column cache can only prove single-column order
    if sel.len() != 1 {
        return false;
    }
    let Some(s) = stats.get(sel[0]) else {
        return false;
    };

    // empty/NULL cells are excluded from the cache's sort_order computation, but
    // the real comparators sort them as Less - so any nulls void the proof
    if s.nullcount != 0 {
        return false;
    }

    // all-null columns have no sort_order
    if s.sort_order.as_deref() != Some("Ascending") {
        return false;
    }

    // the cache's comparison semantics are type-dependent; they must match what
    // the caller will actually do
    if requested_numeric {
        matches!(s.r#type.as_str(), "Integer" | "Float")
    } else {
        s.r#type == "String" && (!for_extsort || s.is_ascii)
    }
}

/// Reads a CSV file (comma-delimited) and writes it with a different delimiter to a writer.
pub fn csv_to_delimited_writer<W: Write>(
    input_csv: &str,
    writer: &mut W,
    delimiter: u8,
) -> CliResult<()> {
    let file = File::open(input_csv)?;
    let mut rdr = csv::ReaderBuilder::new()
        .has_headers(true)
        .delimiter(b',')
        .from_reader(file);
    let mut wtr = csv::WriterBuilder::new()
        .delimiter(delimiter)
        .from_writer(writer);

    let headers = rdr.headers()?.clone();
    wtr.write_record(&headers)?;

    for result in rdr.records() {
        let record = result?;
        wtr.write_record(&record)?;
    }
    wtr.flush()?;
    Ok(())
}

pub fn csv_to_jsonl(
    input_csv: &str,
    csv_types: &phf::Map<&'static str, JsonTypes>,
    output_jsonl: &PathBuf,
    delimiter: u8,
) -> CliResult<()> {
    let output = File::create(output_jsonl)?;
    let writer = BufWriter::new(output);
    csv_to_jsonl_writer(input_csv, csv_types, writer, delimiter)
}

// Build a `serde_json::Map` from a CSV record, coercing each cell to the JSON type
// declared in `csv_types` (defaulting to String for unknown keys). Empty cells are
// skipped entirely (no key is emitted), matching the JSONL sidecar behavior.
fn csv_record_to_json_map(
    record: &csv::StringRecord,
    key_vec: &[String],
    csv_types: &phf::Map<&'static str, JsonTypes>,
    json_object: &mut serde_json::Map<String, serde_json::Value>,
) {
    json_object.clear();
    for (i, val) in record.iter().enumerate() {
        // Skip extra fields when a record has more columns than headers.
        let Some(key) = key_vec.get(i) else {
            break;
        };
        let data_type = csv_types.get(key.as_str()).unwrap_or(&JsonTypes::String);
        let value = if val.is_empty() {
            continue;
        } else {
            match *data_type {
                JsonTypes::String => serde_json::Value::String(val.to_owned()),
                JsonTypes::Int => {
                    if let Ok(num) = val.parse::<u64>() {
                        serde_json::Value::Number(serde_json::Number::from(num))
                    } else {
                        serde_json::Value::String(val.to_owned())
                    }
                },
                JsonTypes::Float => {
                    if let Ok(num) = val.parse::<f64>() {
                        if let Some(n) = serde_json::Number::from_f64(num) {
                            serde_json::Value::Number(n)
                        } else {
                            serde_json::Value::Number(
                                serde_json::Number::from_f64(0.0).unwrap_or_else(|| {
                                    // safety: we know that 0.0 is a valid f64
                                    serde_json::Number::from_f64(0.0).unwrap()
                                }),
                            )
                        }
                    } else {
                        serde_json::Value::Number(
                            serde_json::Number::from_f64(0.0)
                                // safety: we know that 0.0 is a valid f64
                                .unwrap_or_else(|| serde_json::Number::from_f64(0.0).unwrap()),
                        )
                    }
                },
                JsonTypes::Bool => serde_json::Value::Bool(val.parse::<bool>().unwrap_or(false)),
            }
        };
        json_object.insert(key.to_string(), value);
    }
}

// Stream a CSV file to a writer as JSON Lines (NDJSON) - one JSON object per record.
// This is the shared core of `csv_to_jsonl` (which writes to a file) and the
// `qsv stats --jsonl` stdout output mode.
pub fn csv_to_jsonl_writer<W: Write>(
    input_csv: &str,
    csv_types: &phf::Map<&'static str, JsonTypes>,
    mut writer: W,
    delimiter: u8,
) -> CliResult<()> {
    let file = File::open(input_csv)?;
    let mut rdr = csv::ReaderBuilder::new()
        .has_headers(true)
        .delimiter(delimiter)
        .from_reader(file);

    let headers = rdr.headers()?;
    let key_vec: Vec<String> = headers
        .iter()
        .map(std::string::ToString::to_string)
        .collect();

    let mut json_object = serde_json::Map::with_capacity(key_vec.len());
    let mut record = csv::StringRecord::new();
    let mut json_line: String;

    while rdr.read_record(&mut record)? {
        csv_record_to_json_map(&record, &key_vec, csv_types, &mut json_object);

        // Use platform-appropriate JSON serialization
        json_line = cfg_select! {
            target_endian = "little" => simd_json::to_string(&json_object)?,
            _ => serde_json::to_string(&json_object)?,
        };
        writeln!(writer, "{json_line}")?;
    }

    Ok(writer.flush()?)
}

// Write a CSV file to a writer as a single pretty-printed JSON array of per-record
// objects. Used by the `qsv stats --pretty-json` stdout output mode.
pub fn csv_to_json_array_writer<W: Write>(
    input_csv: &str,
    csv_types: &phf::Map<&'static str, JsonTypes>,
    mut writer: W,
    delimiter: u8,
) -> CliResult<()> {
    let file = File::open(input_csv)?;
    let mut rdr = csv::ReaderBuilder::new()
        .has_headers(true)
        .delimiter(delimiter)
        .from_reader(file);

    let headers = rdr.headers()?;
    let key_vec: Vec<String> = headers
        .iter()
        .map(std::string::ToString::to_string)
        .collect();

    let mut json_object = serde_json::Map::with_capacity(key_vec.len());
    let mut record = csv::StringRecord::new();
    let mut json_array: Vec<serde_json::Value> = Vec::new();

    while rdr.read_record(&mut record)? {
        csv_record_to_json_map(&record, &key_vec, csv_types, &mut json_object);
        json_array.push(serde_json::Value::Object(json_object.clone()));
    }

    // Use platform-appropriate JSON serialization
    let json_output = cfg_select! {
        target_endian = "little" => simd_json::to_string_pretty(&json_array)?,
        _ => serde_json::to_string_pretty(&json_array)?,
    };
    writeln!(writer, "{json_output}")?;

    Ok(writer.flush()?)
}

/// get the optimal batch size
/// if CSV is not indexed and `ROW_COUNT` is not set, return `DEFAULT_BATCH_SIZE`
/// if `batch_size` is 0, return the number of rows in the CSV, effectively disabling batching
/// if `batch_size` is 1, force `batch_size` to be set to "`optimal_size`", even though
/// its not recommended (number of rows is too small for parallel processing)
/// if `batch_size` is equal to `DEFAULT_BATCH_SIZE`, return the `optimal_size`
/// failing everything above, return the requested `batch_size`
#[inline]
pub fn optimal_batch_size(rconfig: &Config, batch_size: usize, num_jobs: usize) -> usize {
    if batch_size > 1 && batch_size < DEFAULT_BATCH_SIZE {
        return DEFAULT_BATCH_SIZE;
    }

    let num_rows = match ROW_COUNT.get() {
        // ROW_COUNT is OnceLock<Option<u64>>; the inner Option is None when
        // count_rows_with_best_method failed (both polars and the CSV reader).
        // Fall through to the indexed path in that case rather than panicking.
        Some(Some(count)) => *count as usize,
        _ => match rconfig.indexed() {
            Ok(Some(idx)) => idx.count() as usize,
            _ => {
                return DEFAULT_BATCH_SIZE;
            },
        },
    };

    if batch_size == 0 {
        // disable batching, handle all rows in one batch
        num_rows
    } else if (num_rows > DEFAULT_BATCH_SIZE && (batch_size == DEFAULT_BATCH_SIZE))
        || batch_size == 1
    {
        // the optimal batch size is the number of rows divided by the number of jobs
        if num_rows.is_multiple_of(num_jobs) {
            // there is no remainder as num_rows is divisible by num_jobs
            num_rows / num_jobs
        } else {
            // there is a remainder, we add 1 to the batch size
            // this is to ensure that all rows are processed
            (num_rows / num_jobs) + 1
        }
    } else {
        batch_size
    }
}

/// Expand the tilde (`~`) from within the provided path.
// consumed by non-lite commands (describegpt, fetchpost, geocode) and the
// `get`/diskcache subsystem (none of which are in qsvlite), plus the throttled
// self-update check - which can be enabled alongside `lite`, hence the extra gate.
#[cfg(any(not(feature = "lite"), feature = "self_update"))]
pub fn expand_tilde(path: impl AsRef<Path>) -> Option<PathBuf> {
    let p = path.as_ref();

    let expanded = if p.starts_with("~") {
        let mut base = directories::BaseDirs::new()?.home_dir().to_path_buf();

        if !p.ends_with("~") {
            base.extend(p.components().skip(1));
        }
        base
    } else {
        p.to_path_buf()
    };
    Some(expanded)
}

// comment out for now as this is still WIP
// pub fn create_json_record(
//     no_headers: bool,
//     headers: &csv::ByteRecord,
//     record: &csv::ByteRecord,
//     is_first: &mut bool,
// ) -> CliResult<String> {
//     let header_vec: Vec<String> = headers
//         .iter()
//         .enumerate()
//         .map(|(col_idx, b)| {
//             if no_headers {
//                 col_idx.to_string()
//             } else {
//                 String::from_utf8_lossy(b).to_string()
//             }
//         })
//         .collect();

//     let mut json_record = String::new();

//     let rec_len = header_vec.len().saturating_sub(1);
//     let mut temp_val;
//     let mut json_string_val: serde_json::Value;
//     let null_val = "null".to_string();

//     if *is_first {
//         // write!(json_wtr, "{{")?;
//         json_record.push('{');
//         *is_first = false;
//     } else {
//         // write!(json_wtr, ",{{")?;
//         json_record.push_str(",{");
//     }
//     for (idx, b) in record.iter().enumerate() {
//         if let Ok(val) = simdutf8::basic::from_utf8(b) {
//             temp_val = val.to_owned();
//         } else {
//             temp_val = String::from_utf8_lossy(b).to_string();
//         }
//         if temp_val.is_empty() {
//             temp_val.clone_from(&null_val);
//         } else {
//             json_string_val = serde_json::Value::String(temp_val);
//             temp_val = json_string_val.to_string();
//         }
//         if idx < rec_len {
//             unsafe {
//                 // write!(
//                 //     json_wtr,
//                 //     r#""{key}":{value},"#,
//                 //     key = header_vec.get_unchecked(idx),
//                 //     value = temp_val
//                 // )?;
//                 json_record.push_str(&format!(
//                     r#""{key}":{value},"#,
//                     key = header_vec.get_unchecked(idx),
//                     value = temp_val
//                 ));
//             }
//         } else {
//             unsafe {
//                 // write!(
//                 //     json_wtr,
//                 //     r#""{key}":{value}"#,
//                 //     key = header_vec.get_unchecked(idx),
//                 //     value = temp_val
//                 // )?;
//                 json_record.push_str(&format!(
//                     r#""{key}":{value}"#,
//                     key = header_vec.get_unchecked(idx),
//                     value = temp_val
//                 ));
//             }
//         }
//     }
//     // Ok(write!(json_wtr, "}}")?)
//     json_record.push('}');
//     Ok(json_record)
// }

/// Loads a Polars schema from a pschema.json file if it exists.
///
/// # Arguments
///
/// * `path` - The path to the input file
///
/// # Returns
///
/// * `Option<Arc<Schema>>` - The loaded schema if the file exists and can be parsed, None otherwise
#[cfg(feature = "polars")]
fn load_schema_from_file(path: &Path) -> Result<Option<Arc<Schema>>, Box<dyn std::error::Error>> {
    // Append .pschema.json to the full filename (including extension)
    // e.g. data.csv -> data.csv.pschema.json, data.tsv.gz -> data.tsv.gz.pschema.json
    let schema_file = PathBuf::from(format!("{}.pschema.json", path.display()));

    if schema_file.exists() {
        // Load the schema from the pschema.json file
        let file = File::open(&schema_file)?;
        let mut buf_reader = BufReader::new(file);
        let mut schema_json = String::with_capacity(100);
        buf_reader.read_to_string(&mut schema_json)?;
        let schema: Schema = serde_json::from_str(&schema_json)?;
        Ok(Some(Arc::new(schema)))
    } else {
        Ok(None)
    }
}

/// Tabular file extensions qsv recognizes for delimiter inference inside a zip.
const ZIP_TABULAR_EXTS: [&str; 4] = ["csv", "tsv", "tab", "ssv"];

/// Choose which entry of a zip archive to treat as the tabular payload.
///
/// Returns the chosen entry's 0-based index and its lower-cased tabular extension
/// (`csv`/`tsv`/`tab`/`ssv`). Selection rule:
/// 1. Skip directory entries and system files (e.g. `__MACOSX`, `.DS_Store`) via
///    `root_dir_common_filter`.
/// 2. Use the first entry (in archive order) whose extension is tabular.
/// 3. Error if no tabular entry exists — including the single-entry case. We do NOT silently treat
///    an arbitrary file as CSV, matching the documented "first CSV/TSV/TAB/SSV entry" behavior.
///
/// By design, a `.zip` is treated as a container of delimited text. A **special
/// binary format** (Parquet/Avro/Arrow) nested inside a `.zip` is intentionally
/// NOT selectable here — those formats are already compressed, so zipping them is
/// not an expected workflow; provide such files directly instead. (Issue #3988 —
/// "nested special formats": won't-support, documented in README.)
///
/// Generic over `Read + Seek` so it serves both file-backed archives
/// (`extract_zip_to_temp`) and in-memory `Cursor`-backed ones (`diskcache`).
pub fn select_zip_entry<R: Read + std::io::Seek>(
    archive: &mut zip::ZipArchive<R>,
) -> CliResult<(usize, Option<String>)> {
    let mut usable_count = 0_usize;
    let mut first_tabular: Option<(usize, String)> = None;

    for i in 0..archive.len() {
        // Inspect the central-directory name only (no decoder is constructed), so
        // an encrypted or unsupported-compression entry elsewhere in the archive
        // can't fail selection of a valid earlier CSV/TSV/TAB/SSV entry. The chosen
        // entry is decoded later (by `extract_zip_to_temp` via `by_index`).
        let entry = archive.by_index_raw(i)?;
        let name = entry.name().to_string();
        if name.ends_with('/') || !root_dir_common_filter(Path::new(&name)) {
            continue;
        }
        usable_count += 1;
        if first_tabular.is_none()
            && let Some(ext) = Path::new(&name)
                .extension()
                .and_then(|e| e.to_str())
                .map(str::to_ascii_lowercase)
                .filter(|e| ZIP_TABULAR_EXTS.contains(&e.as_str()))
        {
            first_tabular = Some((i, ext));
        }
    }

    if let Some((idx, ext)) = first_tabular {
        if usable_count > 1 {
            log::info!(
                "zip archive has multiple entries; using first CSV/TSV/TAB/SSV entry (index {idx})"
            );
        }
        return Ok((idx, Some(ext)));
    }
    Err(CliError::IncorrectUsage(if usable_count == 0 {
        "zip archive contains no usable file entry.".to_string()
    } else {
        "zip archive contains no CSV/TSV/TAB/SSV entry.".to_string()
    }))
}

/// Map a chosen zip entry's tabular extension to a temp-file suffix that qsv's
/// delimiter inference (`get_delim_by_extension`) understands.
fn zip_inner_ext_to_suffix(inner_ext: Option<&str>) -> &'static str {
    match inner_ext {
        Some("tsv" | "tab") => ".tsv",
        Some("ssv") => ".ssv",
        _ => ".csv",
    }
}

/// Extract the tabular entry of a zip archive to a temp file whose extension
/// selects the delimiter downstream (via `get_delim_by_extension`). The temp file
/// lives in `TEMP_FILE_DIR` and is `keep()`-ed (process-lifetime), matching
/// `convert_special_format`'s temp lifecycle.
///
/// Always compiled (needs only the non-optional `zip` crate), so it is usable in
/// non-polars builds and from `convert_special_format`'s non-polars stub.
pub fn extract_zip_to_temp(
    path: &Path,
    _default_delim: u8,
) -> Result<PathBuf, Box<dyn std::error::Error>> {
    let zip_file = File::open(path)?;
    let mut archive = zip::ZipArchive::new(zip_file)?;
    let (idx, inner_ext) = select_zip_entry(&mut archive).map_err(|e| e.to_string())?;
    let suffix = zip_inner_ext_to_suffix(inner_ext.as_deref());

    let temp_dir =
        crate::config::TEMP_FILE_DIR.get_or_init(|| tempfile::TempDir::new().unwrap().keep());
    let mut temp_file = tempfile::Builder::new()
        .suffix(suffix)
        .tempfile_in(temp_dir)?;

    let mut entry = archive.by_index(idx)?;
    std::io::copy(&mut entry, &mut temp_file)?;
    temp_file.flush()?;

    let out = temp_file.path().to_path_buf();
    temp_file.keep()?;
    Ok(out)
}

/// Extract a zip archive's usable entries into a temp subdirectory and return
/// their paths, **tabular entries first** (CSV/TSV/TAB/SSV in archive order),
/// followed by other supported entries (special formats parquet/avro/json/…, in
/// archive order — or *any* file when `QSV_SKIP_FORMAT_CHECK` is set, via
/// `is_supported_file`). Directories, system files (`__MACOSX`, `.DS_Store`, …),
/// zip-slip entries, and unsupported file types are skipped. Errors with a clear
/// message if no usable entry remains.
///
/// This is the command-level ("extract all entries") counterpart to the
/// reader-level `select_zip_entry`/`extract_zip_to_temp` (which lazily extracts
/// only the first tabular entry). Both share the same selection rule — the
/// tabular extensions in `ZIP_TABULAR_EXTS`, filtered by `root_dir_common_filter`
/// — so a single-input command that consumes the first returned path selects the
/// SAME entry the `Config` reader path would, while multi-input commands (`cat`,
/// `sqlp`, …) still receive every entry. Issue #3988.
fn extract_all_zip_entries(
    path: &Path,
    tmpdir: &tempfile::TempDir,
) -> Result<Vec<PathBuf>, CliError> {
    // A per-archive subdirectory keeps each entry's relative path (zips may carry
    // nested directories) and avoids name collisions across multiple zip inputs.
    let zip_stem = path
        .file_name()
        .and_then(std::ffi::OsStr::to_str)
        .map_or_else(
            || "zip".to_string(),
            |n| {
                n.strip_suffix(".zip")
                    .or_else(|| n.strip_suffix(".ZIP"))
                    .unwrap_or(n)
                    .to_string()
            },
        );
    let zip_extract_dir = tmpdir.path().join(&zip_stem);
    std::fs::create_dir_all(&zip_extract_dir)?;

    let mut archive = zip::ZipArchive::new(File::open(path)?)?;

    // `get_special_format` classifies by inspecting the on-disk file, so each
    // entry must be extracted before it can be classified. Collect tabular and
    // other-supported entries separately to preserve the tabular-first ordering.
    let mut tabular = Vec::new();
    let mut other_supported = Vec::new();
    for i in 0..archive.len() {
        let mut zip_entry = archive.by_index(i)?;
        let entry_name = zip_entry.name().to_string();

        if entry_name.ends_with('/') || !root_dir_common_filter(Path::new(&entry_name)) {
            log::info!("  Skipping system file or directory: {entry_name}");
            continue;
        }

        // Reject path-traversal entries (zip-slip): `enclosed_name()` returns None
        // when the entry name would escape the extraction root.
        let Some(safe_name) = zip_entry.enclosed_name() else {
            log::warn!("  Skipping zip entry with unsafe path (zip-slip): {entry_name}");
            continue;
        };

        let file_path = zip_extract_dir.join(safe_name);
        if let Some(parent) = file_path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        let mut outfile = File::create(&file_path)?;
        std::io::copy(&mut zip_entry, &mut outfile)?;
        outfile.flush()?;
        // Release the write handle before classifying (Windows: a reader can't
        // re-open the path while a write handle is live).
        drop(outfile);

        let ext = file_path
            .extension()
            .and_then(std::ffi::OsStr::to_str)
            .map(str::to_ascii_lowercase)
            .unwrap_or_default();
        if ZIP_TABULAR_EXTS.contains(&ext.as_str()) {
            tabular.push(file_path);
        } else if is_supported_file(&file_path) {
            // Non-tabular but supported: a special format (parquet/avro/json/…),
            // or — when `QSV_SKIP_FORMAT_CHECK` is set — any file (the global
            // escape hatch, honored via `is_supported_file`).
            other_supported.push(file_path);
        } else {
            log::info!("  Skipping unsupported file type: {}", file_path.display());
        }
    }

    if tabular.is_empty() && other_supported.is_empty() {
        return fail_clierror!(
            "zip archive '{}' contains no supported (CSV/TSV/TAB/SSV or special-format) file \
             entry.",
            path.display()
        );
    }

    // Tabular entries first, so a single-input consumer's first element matches
    // the reader-level `select_zip_entry` choice (the first CSV/TSV/TAB/SSV entry).
    tabular.extend(other_supported);
    Ok(tabular)
}

/// Converts files in special formats (Parquet, Avro, Arrow IPC, JSONL, JSON, or compressed CSV)
/// into a standard delimited text file. The output file extension will be:
/// - .tsv for tab-delimited
/// - .ssv for semicolon-delimited
/// - .csv for comma-delimited
///
/// # Arguments
///
/// * `path` - The path to the input file.
/// * `format` - The format of the input file.
/// * `delim` - The delimiter to use for the output CSV file.
///
/// # Returns
///
/// A `Result` containing the path to the temporary CSV file.
/// The caller is responsible for deleting the temporary file.
#[cfg(feature = "polars")]
pub fn convert_special_format(
    path: &Path,
    format: SpecialFormat,
    delim: u8,
) -> Result<PathBuf, Box<dyn std::error::Error>> {
    use polars::{
        io::avro::AvroReader,
        prelude::{
            CsvParseOptions, CsvReadOptions, CsvWriter, IpcReader, JsonReader, LazyFileListReader,
            LazyJsonLineReader, ParquetReader, PlRefPath, SerReader, SerWriter,
        },
    };

    // Zip is not a polars-native format; extract the chosen tabular entry to a
    // temp file (always-compiled path) and return it directly.
    if format == SpecialFormat::CompressedZip {
        return extract_zip_to_temp(path, delim);
    }

    // Check if there's a pschema.json file with the same filestem
    // the Polars schema will be used in parsing
    // JSON/JSONL and compressed CSV files only
    let schema = if let SpecialFormat::Avro | SpecialFormat::Parquet | SpecialFormat::Ipc = format {
        None
    } else {
        load_schema_from_file(path)?
    };

    let mut extension = ".csv";
    // Create a reader based on the file format and convert to DataFrame
    let mut df = match format {
        SpecialFormat::Avro => AvroReader::new(BufReader::new(File::open(path)?)).finish()?,
        SpecialFormat::Parquet => ParquetReader::new(BufReader::new(File::open(path)?)).finish()?,
        SpecialFormat::Ipc => IpcReader::new(BufReader::new(File::open(path)?)).finish()?,
        SpecialFormat::Jsonl => {
            let path_str = path.to_string_lossy();
            let lf = LazyJsonLineReader::new(PlRefPath::new(&*path_str));
            if let Some(schema) = schema {
                lf.with_schema(Some(schema)).finish()?
            } else {
                lf.finish()?
            }
            .collect()?
        },
        SpecialFormat::Json => {
            let df = JsonReader::new(BufReader::new(File::open(path)?));
            if let Some(schema) = schema {
                df.with_schema(schema).finish()?
            } else {
                df.finish()?
            }
        },
        SpecialFormat::CompressedCsv
        | SpecialFormat::CompressedTsv
        | SpecialFormat::CompressedSsv => {
            let separator = match format {
                SpecialFormat::CompressedTsv => {
                    extension = ".tsv";
                    b'\t'
                },
                SpecialFormat::CompressedSsv => {
                    extension = ".ssv";
                    b';'
                },
                _ => delim,
            };

            // Create base CSV read options with the appropriate separator
            let base_options = CsvReadOptions::default()
                .with_parse_options(CsvParseOptions::default().with_separator(separator));

            // Try reading the compressed file with a schema if available
            let reader = CsvReadOptions::default()
                .try_into_reader_with_file_path(Some(path.to_path_buf()))?
                .with_options(if let Some(schema) = schema {
                    base_options.clone().with_schema(Some(schema))
                } else {
                    // it failed, try to infer it with 1,000 rows
                    base_options.clone().with_infer_schema_length(Some(1_000))
                });

            if let Ok(df) = reader.finish() {
                df
            } else {
                // Got an error. Try again with a larger infer schema length of 10,000 rows
                log::warn!(
                    "Falling back to reading file \"{}\" without a schema. 2nd try using infer \
                     schema length of 10,000 rows.",
                    path.display()
                );

                let reader_2ndtry = CsvReadOptions::default()
                    .try_into_reader_with_file_path(Some(path.to_path_buf()))?
                    .with_options(base_options.clone().with_infer_schema_length(Some(10_000)));

                if let Ok(df) = reader_2ndtry.finish() {
                    df
                } else {
                    log::warn!("Still failing. 3rd try - scanning the whole file to infer schema.");

                    // Try one last time without an infer schema length, scanning the whole file
                    let reader_3rdtry = CsvReadOptions::default()
                        .try_into_reader_with_file_path(Some(path.to_path_buf()))?
                        .with_options(base_options.with_infer_schema_length(None));

                    reader_3rdtry.finish()?
                }
            }
        },
        SpecialFormat::Unknown => return Err("Unknown format".into()),
        // handled by the early return at the top of this function
        SpecialFormat::CompressedZip => unreachable!(),
    };

    // Get or initialize temp directory that persists until program exit
    // safety: we know that the tempfile::TempDir::new() will not ordinarily fail
    // otherwise, we have a bigger problem
    let temp_dir =
        crate::config::TEMP_FILE_DIR.get_or_init(|| tempfile::TempDir::new().unwrap().keep());

    // Create temp file with appropriate extension
    let mut temp_file = tempfile::Builder::new()
        .suffix(extension)
        .tempfile_in(temp_dir)?;

    // Get QSV_POLARS_FORMAT_FLOAT_PRECISION env var
    let precision = crate::config::POLARS_FLOAT_PRECISION.get_or_init(|| {
        std::env::var("QSV_POLARS_FLOAT_PRECISION")
            .ok()
            .and_then(|s| s.parse().ok())
    });

    // Write DataFrame to CSV with specified delimiter/separator
    CsvWriter::new(BufWriter::new(&temp_file))
        .with_separator(delim)
        .with_float_precision(*precision)
        .finish(&mut df)?;
    temp_file.flush()?;

    let path = temp_file.path().to_path_buf();
    temp_file.keep()?; // Prevent auto-deletion

    Ok(path)
}

#[cfg(not(feature = "polars"))]
#[allow(unused_variables)]
pub fn convert_special_format(
    path: &Path,
    format: SpecialFormat,
    delim: u8,
) -> Result<PathBuf, Box<dyn std::error::Error>> {
    // Zip extraction needs only the always-compiled `zip` crate, so it works
    // even in non-polars builds.
    if format == SpecialFormat::CompressedZip {
        return extract_zip_to_temp(path, delim);
    }
    Err(
        "This file type cannot be opened with your current version of qsv. You need the full, \
         polars-enabled version to work with Avro, Arrow, Parquet, JSON/JSONL and gzip/zlib/zst \
         compressed files. Please download the full version from the qsv website."
            .into(),
    )
}

#[cfg(feature = "polars")]
pub fn infer_polars_schema(
    delimiter: Option<crate::config::Delimiter>,
    debuglog_flag: bool,
    table: &Path,
    schema_file: &std::path::PathBuf,
    prefer_dmy: bool,
) -> Result<bool, crate::clitypes::CliError> {
    // Run sniff to auto-detect date/datetime columns
    let qsv_bin = current_exe()?;
    let mut sniff_cmd = std::process::Command::new(&qsv_bin);
    let table_str = table.to_string_lossy().to_string();
    sniff_cmd.args(["sniff", "--json"]);
    if let Some(d) = delimiter {
        let delim_byte = d.as_byte();
        if delim_byte == b'\t' {
            sniff_cmd.args(["--delimiter", r"\t"]);
        } else {
            // safety: delimiter is guaranteed to be a valid ASCII byte
            sniff_cmd.args(["--delimiter", &(delim_byte as char).to_string()]);
        }
    }
    sniff_cmd.arg(&table_str);
    if prefer_dmy {
        sniff_cmd.arg("--prefer-dmy");
    }
    let sniff_output = sniff_cmd.output()?;

    let mut sniff_dates_whitelist = String::new();
    if sniff_output.status.success() {
        match serde_json::from_slice::<serde_json::Value>(&sniff_output.stdout) {
            Ok(sniff_json) => {
                if let (Some(fields), Some(types)) = (
                    sniff_json["fields"].as_array(),
                    sniff_json["types"].as_array(),
                ) {
                    let date_fields: Vec<&str> = fields
                        .iter()
                        .zip(types.iter())
                        .filter_map(|(field, typ)| {
                            let type_str = typ.as_str().unwrap_or_default();
                            if type_str == "Date" || type_str == "DateTime" {
                                field.as_str()
                            } else {
                                None
                            }
                        })
                        .collect();
                    sniff_dates_whitelist = date_fields.join(",");
                    if debuglog_flag {
                        if sniff_dates_whitelist.is_empty() {
                            log::debug!(
                                "sniff did not detect any date/datetime columns for {table_str}"
                            );
                        } else {
                            log::debug!(
                                "sniff detected date/datetime columns for {table_str}: \
                                 {sniff_dates_whitelist}"
                            );
                        }
                    }
                } else if debuglog_flag {
                    log::debug!(
                        "sniff JSON for {table_str} did not contain expected 'fields'/'types' \
                         arrays"
                    );
                }
            },
            Err(err) => {
                if debuglog_flag {
                    log::debug!("failed to parse sniff JSON output for {table_str}: {err}");
                }
            },
        }
    } else if debuglog_flag {
        log::debug!(
            "sniff command failed for {table_str} with status {:?}. stderr: {}",
            sniff_output.status,
            String::from_utf8_lossy(&sniff_output.stderr)
        );
    }

    let schema_args = SchemaArgs {
        flag_enum_threshold:  0,
        flag_ignore_case:     false,
        flag_strict_dates:    false,
        flag_strict_formats:  false,
        // we still get all the stats columns so we can use the stats cache
        flag_pattern_columns: crate::select::SelectColumns::parse("").unwrap(),
        flag_dates_whitelist: sniff_dates_whitelist,
        flag_prefer_dmy:      prefer_dmy,
        flag_force:           false,
        flag_stdout:          false,
        flag_jobs:            Some(njobs(None)),
        flag_polars:          false,
        flag_no_headers:      false,
        flag_delimiter:       delimiter,
        arg_input:            Some(table.to_string_lossy().into_owned()),
        flag_memcheck:        false,
        flag_output:          None,
    };
    let (csv_fields, csv_stats) = get_stats_records(&schema_args, StatsMode::PolarsSchema)?;
    let mut schema = polars::prelude::Schema::with_capacity(csv_stats.len());

    // fetch the decimal scale from the QSV_POLARS_DECIMAL_SCALE env var
    let scale = std::env::var("QSV_POLARS_DECIMAL_SCALE")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(5); // default scale is 5
    for (idx, stat) in csv_stats.iter().enumerate() {
        // safety: we know that the get(idx) will not be None as we are using an iterator
        schema.insert(
            polars::prelude::PlSmallStr::from_str(
                simdutf8::basic::from_utf8(csv_fields.get(idx).unwrap()).unwrap(),
            ),
            {
                let datatype = &stat.r#type;
                #[allow(clippy::match_same_arms)]
                match datatype.as_str() {
                    "String" => polars::datatypes::DataType::String,
                    "Integer" => {
                        // safety: integer types are guaranteed to have a min and max
                        let min = stat.min.as_ref().unwrap();
                        let max = stat.max.as_ref().unwrap();

                        // Check if all values are non-negative to
                        // use unsigned types
                        if let (Ok(min_val), Ok(max_val)) = (min.parse::<i64>(), max.parse::<i64>())
                        {
                            if min_val >= 0 {
                                // Use smallest unsigned type that can hold
                                // the max value
                                if max_val <= u8::MAX as i64 {
                                    polars::datatypes::DataType::UInt8
                                } else if max_val <= u16::MAX as i64 {
                                    polars::datatypes::DataType::UInt16
                                } else if max_val <= u32::MAX as i64 {
                                    polars::datatypes::DataType::UInt32
                                } else {
                                    polars::datatypes::DataType::UInt64
                                }
                            } else {
                                // Use signed types for negative values
                                if min_val >= i32::MIN as i64 && max_val <= i32::MAX as i64 {
                                    polars::datatypes::DataType::Int32
                                } else {
                                    polars::datatypes::DataType::Int64
                                }
                            }
                        } else {
                            // Fallback to Int64 if parsing fails
                            polars::datatypes::DataType::Int64
                        }
                    },
                    "Float" => {
                        // safety: float types are guaranteed to have a min and max
                        let min = stat.min.as_ref().unwrap();
                        let max = stat.max.as_ref().unwrap();
                        let precision = stat.max_precision.unwrap_or(0);

                        // As we use f64 internally, its unlikely that we have more
                        // than 16 digits of precision, but we do this anyway to
                        // document it as the polars engine does support it
                        if precision > 16 {
                            // For very high precision, use Decimal type
                            polars::datatypes::DataType::Decimal(precision as usize, scale)
                        } else if precision > 7
                            || min.parse::<f32>().is_err()
                            || max.parse::<f32>().is_err()
                        {
                            polars::datatypes::DataType::Float64
                        } else {
                            polars::datatypes::DataType::Float32
                        }
                    },
                    "Boolean" => polars::datatypes::DataType::Boolean,
                    "Date" => polars::datatypes::DataType::Date,
                    "DateTime" => polars::datatypes::DataType::Datetime(
                        polars::datatypes::TimeUnit::Milliseconds,
                        None,
                    ),
                    _ => polars::datatypes::DataType::String,
                }
            },
        );
    }
    let stats_schema = std::sync::Arc::new(schema);
    // Use serde_json for schema serialization as the schema may contain compound types
    // (e.g. Datetime) that simd_json::to_string_pretty doesn't serialize correctly
    let stats_schema_json = serde_json::to_string_pretty(&stats_schema)?;
    let mut file = std::io::BufWriter::new(File::create(schema_file)?);
    file.write_all(stats_schema_json.as_bytes())?;
    file.flush()?;
    if debuglog_flag {
        log::debug!("Saved stats_schema to file: {}", schema_file.display());
    }
    Ok(true)
}

/// BLAKE3 hash of a file optimized for maximum performance
/// Uses memory mapping and multithreading for fast hashing of files of any size
// only consumed by describegpt, which is excluded from qsvlite
#[cfg(not(feature = "lite"))]
pub fn hash_blake3_file(path: &Path) -> CliResult<String> {
    let mut hasher = blake3::Hasher::new();

    // Use BLAKE3's optimized memory-mapped + rayon parallel hashing
    // This automatically handles chunking and parallel processing internally
    hasher.update_mmap_rayon(path)?;

    Ok(hasher.finalize().to_hex().to_string())
}

#[cfg(not(feature = "lite"))]
pub fn is_executable(path: &str) -> std::io::Result<bool> {
    cfg_select! {
        unix => {
            use std::{fs, os::unix::fs::PermissionsExt};

            let metadata = fs::metadata(path)?;
            Ok(metadata.permissions().mode() & 0o111 != 0)
        },
        windows => {
            use std::path::Path;
            let p = Path::new(path);
            Ok(p.extension().and_then(|e| e.to_str()).is_some_and(|ext| {
                matches!(
                    ext.to_ascii_lowercase().as_str(),
                    "exe" | "bat" | "cmd" | "com"
                )
            }))
        },
    }
}

/// Print a status message with elapsed time if not in quiet mode
pub fn print_status(msg: &str, elapsed: Option<std::time::Duration>) {
    // this checks the QUIET_FLAG atomic boolean if it was set
    // Otherwise, it defaults to false and prints the message
    if !QUIET_FLAG.load(std::sync::atomic::Ordering::Relaxed) {
        if let Some(duration) = elapsed {
            eprintln!("{msg} (elapsed: {:.2}s)", duration.as_secs_f64());
        } else {
            eprintln!("{msg}");
        }
    }
}

/// Get the stats CSV file path for a given input CSV path.
/// Returns an absolute path like `/path/to/input.stats.csv`.
pub fn get_stats_csv_path(input_path: &std::path::Path) -> crate::CliResult<std::path::PathBuf> {
    let parent = input_path
        .parent()
        .unwrap_or_else(|| std::path::Path::new("."));
    let fstem = input_path.file_stem().ok_or_else(|| {
        crate::clitypes::CliError::Other("Invalid input path: no file name".to_string())
    })?;

    let stats_filename = format!("{}.stats.csv", fstem.to_string_lossy());
    let result = parent.join(stats_filename);
    if result.is_absolute() {
        Ok(result)
    } else {
        Ok(std::env::current_dir()?.join(result))
    }
}

// Helper function to run qsv commands with consistent error handling and timing
pub fn run_qsv_cmd(
    command: &str,
    args: &[&str],
    input_path: &str,
    status_msg: &str,
) -> CliResult<(String, String)> {
    let start_time = Instant::now();

    // safety: we know that the current_exe() is very unlikely to fail as qsv is already running
    let qsv_path = QSV_PATH.get_or_init(|| current_exe().unwrap().to_string_lossy().to_string());
    let mut cmd = Command::new(qsv_path);

    // sample takes a positional <sample-size> argument *before* the input path
    // (e.g. `qsv sample 1000 input.csv`), so for sample we order args before
    // input. Every other command in qsv accepts the input as the first
    // positional, so input goes before flags/args.
    if command == "sample" {
        cmd.arg(command).args(args).arg(input_path);
    } else {
        cmd.arg(command).arg(input_path).args(args);
    }

    let output = cmd
        .output()
        .map_err(|e| CliError::Other(format!("Error while executing command {command}: {e:?}")))?;
    log::debug!("qsv command {command} output: {output:?}");

    if !output.status.success() {
        return fail_clierror!("Command {command} failed: {output:?}");
    }

    print_status(status_msg, Some(start_time.elapsed()));

    let stdout_str = std::str::from_utf8(&output.stdout).map_err(|e| {
        CliError::Other(format!(
            "Unable to parse output of qsv command {command}: {e:?}"
        ))
    })?;
    let stderr_str = std::str::from_utf8(&output.stderr).map_err(|e| {
        CliError::Other(format!(
            "Unable to parse stderr of qsv command {command}: {e:?}"
        ))
    })?;

    Ok((stdout_str.to_string(), stderr_str.to_string()))
}

/// Like `run_qsv_cmd`, but *streams* the child's live progress to the terminal instead of capturing
/// its output. The child inherits STDERR, so its own `winfo!`/status lines and progress bars render
/// as it runs; STDOUT is discarded so the child can never corrupt this process's own stdout (e.g.
/// `viz`'s piped HTML). `QSV_PROGRESSBAR=1` is set for the child so its OWN in-process progress
/// bars are enabled. This surfaces only what the child writes to its own stderr — it does NOT
/// reveal progress from any subprocess the child itself spawns and captures (e.g. `moarstats` runs
/// its nested `stats`/`frequency` via capturing calls, so those stay hidden). Returns `Ok(())` on
/// success; use when the caller only needs the child's on-disk side effects (e.g. `moarstats`'
/// stats/bivariate cache sidecars), not its stdout. Callers that render their own progress bar
/// should suspend it around this call so the two don't collide.
pub fn run_qsv_cmd_streaming(
    command: &str,
    args: &[&str],
    input_path: &str,
    status_msg: &str,
) -> CliResult<()> {
    let start_time = Instant::now();

    // safety: we know that the current_exe() is very unlikely to fail as qsv is already running
    let qsv_path = QSV_PATH.get_or_init(|| current_exe().unwrap().to_string_lossy().to_string());
    let mut cmd = Command::new(qsv_path);

    if command == "sample" {
        cmd.arg(command).args(args).arg(input_path);
    } else {
        cmd.arg(command).arg(input_path).args(args);
    }

    cmd.env("QSV_PROGRESSBAR", "1")
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::inherit());

    let status = cmd
        .status()
        .map_err(|e| CliError::Other(format!("Error while executing command {command}: {e:?}")))?;

    if !status.success() {
        return fail_clierror!("Command {command} failed with exit status: {status}");
    }

    print_status(status_msg, Some(start_time.elapsed()));
    Ok(())
}

/// Sync subprocess output to disk and validate it is non-empty.
///
/// After a child `qsv` subprocess writes its output via `--output <path>` and
/// exits, the file's bytes may still be in the kernel page cache and not yet
/// visible to follow-up `open()` calls in another process. This is especially
/// noticeable on macOS APFS under heavy parallel test load and has produced
/// silent "primary-only" join output in CI.
///
/// Opens the file with WRITE access (no truncate, no create — just a writable
/// handle), calls `sync_all()` to force a flush, then verifies the resulting
/// file is non-empty.
///
/// Why write access: on Windows, `sync_all()` is implemented via
/// `FlushFileBuffers`, which rejects read-only handles with "Access is
/// denied" (os error 5). Opening with `.write(true)` works on Linux/macOS too
/// and doesn't modify the file as long as `truncate`/`create` are not set.
pub fn sync_subprocess_output(path: &Path) -> CliResult<()> {
    let f = std::fs::OpenOptions::new()
        .write(true)
        .open(path)
        .map_err(|e| {
            CliError::Other(format!(
                "Unable to open subprocess output for fsync ({}): {e}",
                path.display()
            ))
        })?;
    f.sync_all().map_err(|e| {
        CliError::Other(format!(
            "Failed to sync subprocess output ({}): {e}",
            path.display()
        ))
    })?;
    drop(f);

    let size = std::fs::metadata(path)
        .map_err(|e| {
            CliError::Other(format!(
                "Unable to stat subprocess output ({}): {e}",
                path.display()
            ))
        })?
        .len();
    if size == 0 {
        return fail_clierror!("Subprocess output is empty after sync: {}", path.display());
    }
    Ok(())
}

/// Best-effort fsync of the directory at `path` so its directory entry
/// metadata is durable.
///
/// `fsync(file)` on Linux does NOT flush the parent directory's metadata.
/// While that primarily matters for crash safety, some FS configurations
/// (overlayfs, fuse, network mounts) and edge-case timing have been seen
/// to affect read-after-write visibility for newly-created files when a
/// follow-up process opens them via path lookup. This helper is layered
/// on top of the file-level `sync_subprocess_output` and exists purely
/// as a belt-and-suspenders guard.
///
/// Errors are intentionally NOT propagated:
/// - Directory fsync is not supported uniformly across Linux filesystems (FAT, some
///   FUSE/overlay/network mounts return `EINVAL`/`EOPNOTSUPP`).
/// - Even when supported, the caller has already gone through the file-level fsync; failing the
///   whole operation because the directory fsync errored would punish environments that don't
///   support it.
///
/// Errors are logged at debug level. No-op on non-unix targets (on Windows,
/// `FlushFileBuffers` does not accept directory handles).
pub fn sync_directory(path: &Path) {
    #[cfg(unix)]
    {
        // Best-effort: directory fsync is not supported uniformly across
        // Linux filesystems (FAT, some FUSE/overlay/network mounts return
        // EINVAL/EOPNOTSUPP). Since this is a defensive guard layered on
        // top of the file-level fsync, an error here must NOT fail the
        // caller — log at debug and continue.
        match std::fs::File::open(path) {
            Ok(f) => {
                if let Err(e) = f.sync_all() {
                    log::debug!(
                        "sync_directory: fsync of {} returned {e} (continuing — not all FS \
                         configurations support directory fsync)",
                        path.display()
                    );
                }
            },
            Err(e) => {
                log::debug!(
                    "sync_directory: open of {} returned {e} (continuing — best-effort guard)",
                    path.display()
                );
            },
        }
    }
    #[cfg(not(unix))]
    {
        let _ = path; // suppress unused-variable warning on Windows
    }
}

#[cfg(test)]
mod tests {
    // only used by the hash_blake3_file tests, which are gated out of qsvlite
    #[cfg(not(feature = "lite"))]
    use std::io::Write;

    #[cfg(not(feature = "lite"))]
    use tempfile::NamedTempFile;

    use super::*;

    // A stats cache is located by input path and validated by mtime, NOT by parsing options, so
    // these two fields are the only thing standing between a `--no-headers` (or differently
    // delimited) cache and a headered consumer silently reading positional field names.
    #[test]
    fn stats_cache_parsing_opts_match_rejects_mismatched_no_headers() {
        let input = Path::new("data.csv");
        let meta = serde_json::json!({"flag_no_headers": true, "flag_delimiter": ","});
        // cache built headerless, consumer wants headers -> reject
        assert!(!stats_cache_parsing_opts_match(&meta, input, false, None));
        // same setting on both sides -> accept
        assert!(stats_cache_parsing_opts_match(&meta, input, true, None));
    }

    #[test]
    fn stats_cache_parsing_opts_match_rejects_mismatched_delimiter() {
        let input = Path::new("data.csv");
        let meta = serde_json::json!({"flag_no_headers": false, "flag_delimiter": ";"});
        // cache built semicolon-delimited, consumer resolves ',' from the .csv extension
        assert!(!stats_cache_parsing_opts_match(&meta, input, false, None));
        assert!(stats_cache_parsing_opts_match(
            &meta,
            input,
            false,
            Some(Delimiter(b';'))
        ));
    }

    #[test]
    fn stats_cache_parsing_opts_match_resolves_unset_delimiter_from_extension() {
        // an unset delimiter on BOTH sides resolves from the file extension, so a cache whose
        // recorded delimiter is empty still matches a consumer that passed none
        let meta = serde_json::json!({"flag_no_headers": false, "flag_delimiter": ""});
        assert!(stats_cache_parsing_opts_match(
            &meta,
            Path::new("data.csv"),
            false,
            None
        ));
        // ... and a .tsv input resolves to tab on both sides
        assert!(stats_cache_parsing_opts_match(
            &meta,
            Path::new("data.tsv"),
            false,
            None
        ));
        // but an explicit comma against a .tsv input is a genuine mismatch
        assert!(!stats_cache_parsing_opts_match(
            &meta,
            Path::new("data.tsv"),
            false,
            Some(Delimiter(b','))
        ));
    }

    #[test]
    fn stats_cache_parsing_opts_conflict_only_on_a_readable_mismatch() {
        let dir = tempfile::tempdir().unwrap();
        let input = dir.path().join("data.csv");
        // No sidecar at all: NOT a conflict. `moarstats` writes the JSONL cache without one, and
        // rejecting those would discard a rich cache and regenerate a leaner one.
        assert!(!stats_cache_parsing_opts_conflict(
            &input, &input, false, None
        ));
        // Present but unparseable: same — decline to judge rather than discard.
        std::fs::write(dir.path().join("data.stats.csv.json"), b"{not json").unwrap();
        assert!(!stats_cache_parsing_opts_conflict(
            &input, &input, false, None
        ));
        // Readable and matching: no conflict.
        std::fs::write(
            dir.path().join("data.stats.csv.json"),
            br#"{"flag_no_headers": false, "flag_delimiter": ","}"#,
        )
        .unwrap();
        assert!(!stats_cache_parsing_opts_conflict(
            &input, &input, false, None
        ));
        // Readable and mismatched: the case this exists to catch.
        std::fs::write(
            dir.path().join("data.stats.csv.json"),
            br#"{"flag_no_headers": true, "flag_delimiter": ","}"#,
        )
        .unwrap();
        assert!(stats_cache_parsing_opts_conflict(
            &input, &input, false, None
        ));
    }

    #[test]
    fn stats_cache_parsing_opts_conflict_sees_the_sidecar_beside_a_symlink() {
        // For a symlinked input the JSONL cache is looked up beside the canonical TARGET, but the
        // `stats` subprocess writes its metadata beside the SYMLINK. Checking only the canonical
        // location finds nothing and waves a mismatched cache through.
        let dir = tempfile::tempdir().unwrap();
        let link = dir.path().join("link.csv");
        let canonical = dir.path().join("target.csv");
        // metadata exists ONLY beside the "symlink" path, and it disagrees on --no-headers
        std::fs::write(
            dir.path().join("link.stats.csv.json"),
            br#"{"flag_no_headers": true, "flag_delimiter": ","}"#,
        )
        .unwrap();
        assert!(stats_cache_parsing_opts_conflict(
            &link, &canonical, false, None
        ));
        // and the agreeing case is still not a conflict
        std::fs::write(
            dir.path().join("link.stats.csv.json"),
            br#"{"flag_no_headers": false, "flag_delimiter": ","}"#,
        )
        .unwrap();
        assert!(!stats_cache_parsing_opts_conflict(
            &link, &canonical, false, None
        ));
    }

    #[cfg(test)]
    fn write_test_zip(path: &Path, entries: &[(&str, &[u8])]) {
        use std::io::Write as _;
        let f = std::fs::File::create(path).unwrap();
        let mut zw = zip::ZipWriter::new(f);
        let opts: zip::write::SimpleFileOptions = zip::write::SimpleFileOptions::default();
        for (name, data) in entries {
            zw.start_file(*name, opts).unwrap();
            zw.write_all(data).unwrap();
        }
        zw.finish().unwrap();
    }

    fn sd(r#type: &str, sort_order: Option<&str>, nullcount: u64, is_ascii: bool) -> StatsData {
        StatsData {
            r#type: r#type.to_string(),
            sort_order: sort_order.map(ToString::to_string),
            nullcount,
            is_ascii,
            ..Default::default()
        }
    }

    #[cfg(feature = "self_update")]
    #[test]
    fn update_check_guard_throttles() {
        let dir = tempfile::tempdir().unwrap();
        let stamp = dir.path().join(".last_update_check");
        let guard =
            self_update::check_interval::UpdateCheckGuard::new(&stamp, Duration::from_secs(3600));
        // no stamp yet => a check is due
        assert!(guard.should_check().unwrap());
        // after recording, the interval throttles the next check
        guard.record_check().unwrap();
        assert!(!guard.should_check().unwrap());
    }

    #[test]
    fn proves_ascending_string_lex() {
        let stats = vec![sd("String", Some("Ascending"), 0, true)];
        // default lexicographic, ascending -> hit
        assert!(stats_cache_proves_ascending(&[0], &stats, false, false));
        // extsort variant (ASCII) -> hit
        assert!(stats_cache_proves_ascending(&[0], &stats, false, true));
        // numeric requested on a String column -> miss
        assert!(!stats_cache_proves_ascending(&[0], &stats, true, false));
    }

    #[test]
    fn proves_ascending_descending_never() {
        // a Descending cache is never an ascending proof (--reverse is not
        // short-circuited regardless of direction)
        let stats = vec![sd("String", Some("Descending"), 0, true)];
        assert!(!stats_cache_proves_ascending(&[0], &stats, false, false));
        assert!(!stats_cache_proves_ascending(&[0], &stats, false, true));
    }

    #[test]
    fn proves_ascending_numeric() {
        for typ in ["Integer", "Float"] {
            let stats = vec![sd(typ, Some("Ascending"), 0, true)];
            assert!(stats_cache_proves_ascending(&[0], &stats, true, false));
            // lexicographic requested on a numeric column -> miss
            assert!(!stats_cache_proves_ascending(&[0], &stats, false, false));
        }
    }

    #[test]
    fn proves_ascending_extsort_non_ascii() {
        // ascending String but not ASCII: safe for plain lex, NOT for extsort
        let stats = vec![sd("String", Some("Ascending"), 0, false)];
        assert!(stats_cache_proves_ascending(&[0], &stats, false, false));
        assert!(!stats_cache_proves_ascending(&[0], &stats, false, true));
    }

    #[test]
    fn proves_ascending_guards() {
        let asc = sd("String", Some("Ascending"), 0, true);
        // multi-column selection -> never
        let stats = vec![asc.clone(), asc.clone()];
        assert!(!stats_cache_proves_ascending(&[0, 1], &stats, false, false));
        // out-of-bounds index -> never
        assert!(!stats_cache_proves_ascending(&[5], &stats, false, false));
        // nullcount != 0 -> never
        let nulls = vec![sd("String", Some("Ascending"), 1, true)];
        assert!(!stats_cache_proves_ascending(&[0], &nulls, false, false));
        // Unsorted -> never
        let unsorted = vec![sd("String", Some("Unsorted"), 0, true)];
        assert!(!stats_cache_proves_ascending(&[0], &unsorted, false, false));
        // missing sort_order -> never
        let none = vec![sd("String", None, 0, true)];
        assert!(!stats_cache_proves_ascending(&[0], &none, false, false));
        // Date typed -> never (no matching comparator)
        let date = vec![sd("Date", Some("Ascending"), 0, true)];
        assert!(!stats_cache_proves_ascending(&[0], &date, false, false));
        assert!(!stats_cache_proves_ascending(&[0], &date, true, false));
    }

    #[test]
    fn extract_all_zip_entries_orders_tabular_first_and_filters() {
        let src = tempfile::tempdir().unwrap();
        let zip_path = src.path().join("data.zip");
        // Archive order deliberately puts a system file, an unsupported file, and
        // a non-tabular special-format file BEFORE the tabular entries.
        write_test_zip(
            &zip_path,
            &[
                ("__MACOSX/junk", b"x"),    // system file -> skipped
                ("notes.txt", b"hello"),    // unsupported -> skipped
                ("b.json", b"[{\"a\":1}]"), // special-format supported -> "other"
                ("a.tsv", b"a\tb\n1\t2\n"), // tabular
                ("c.csv", b"x,y\n3,4\n"),   // tabular
            ],
        );

        let work = tempfile::tempdir().unwrap();
        let got = extract_all_zip_entries(&zip_path, &work).unwrap();
        let names: Vec<String> = got
            .iter()
            .map(|p| p.file_name().unwrap().to_string_lossy().into_owned())
            .collect();
        // Tabular entries first (in archive order), then other supported entries;
        // the system file and the unsupported .txt are dropped.
        assert_eq!(names, vec!["a.tsv", "c.csv", "b.json"]);
        // And the first element matches what the reader-level select_zip_entry
        // would pick (the first CSV/TSV/TAB/SSV entry).
        assert!(got[0].exists());
    }

    #[test]
    fn extract_all_zip_entries_errors_when_no_supported_entry() {
        let src = tempfile::tempdir().unwrap();
        let zip_path = src.path().join("nope.zip");
        write_test_zip(&zip_path, &[("readme.txt", b"hi"), ("notes.md", b"yo")]);
        let work = tempfile::tempdir().unwrap();
        let err = extract_all_zip_entries(&zip_path, &work).unwrap_err();
        assert!(
            err.to_string().contains("no supported"),
            "unexpected error: {err}"
        );
    }

    #[cfg(unix)]
    #[test]
    fn test_sync_directory_smoke() {
        // sync_directory is best-effort by design: success paths must
        // return normally, and error paths (missing dir, FS that doesn't
        // support directory fsync) must NOT panic or propagate. This
        // smoke test covers both branches.
        let dir = tempfile::TempDir::new().unwrap();

        // Existing directory: should succeed (or silently no-op if the
        // backing FS doesn't support it).
        sync_directory(dir.path());

        // Non-existent path: should still return without panicking;
        // the open() error is logged and swallowed.
        sync_directory(&dir.path().join("definitely_does_not_exist"));
    }

    #[cfg(not(feature = "lite"))]
    #[test]
    fn test_hash_blake3_file() {
        // Create a temporary file with known content
        let mut temp_file = NamedTempFile::new().unwrap();
        let test_content = b"Hello, World! This is a test file for BLAKE3 hashing.";
        temp_file.write_all(test_content).unwrap();
        temp_file.flush().unwrap();

        // Calculate expected hash using blake3 directly
        let mut expected_hasher = blake3::Hasher::new();
        expected_hasher.update(test_content);
        let expected_hash = expected_hasher.finalize().to_hex().to_string();

        // Test our function
        let actual_hash = hash_blake3_file(temp_file.path()).unwrap();

        assert_eq!(actual_hash, expected_hash);
    }

    #[cfg(not(feature = "lite"))]
    #[test]
    fn test_hash_blake3_file_large() {
        // Create a larger test file (1MB)
        let mut temp_file = NamedTempFile::new().unwrap();
        let test_content = b"Large file test content. ".repeat(40000); // ~1MB
        temp_file.write_all(&test_content).unwrap();
        temp_file.flush().unwrap();

        // Calculate expected hash
        let mut expected_hasher = blake3::Hasher::new();
        expected_hasher.update(&test_content);
        let expected_hash = expected_hasher.finalize().to_hex().to_string();

        // Test our function
        let actual_hash = hash_blake3_file(temp_file.path()).unwrap();

        assert_eq!(actual_hash, expected_hash);
    }

    #[cfg(not(feature = "lite"))]
    #[test]
    fn benchmark_hash_blake3_file() {
        // Create a test file for benchmarking
        let mut temp_file = NamedTempFile::new().unwrap();
        let test_content = b"Benchmark test content. ".repeat(100000); // ~2.4MB
        temp_file.write_all(&test_content).unwrap();
        temp_file.flush().unwrap();

        // Benchmark the function
        let start = std::time::Instant::now();
        let hash = hash_blake3_file(temp_file.path()).unwrap();
        let duration = start.elapsed();

        println!("BLAKE3 Hash: {}", hash);
        println!("BLAKE3 Time: {:?}", duration);
        println!("File size: {} bytes", test_content.len());
        println!(
            "BLAKE3 Speed: {:.2} MB/s",
            (test_content.len() as f64 / 1024.0 / 1024.0) / duration.as_secs_f64()
        );
        assert_eq!(
            hash,
            "87b87c9d36ee75bf8cb30940f6bbbeec3e67328190181f8c59c3cbcd6f35228a"
        );
    }

    #[cfg(not(feature = "lite"))]
    #[test]
    fn benchmark_hash_blake3_file_large() {
        // Create a larger test file (100MB) to test parallel processing
        let mut temp_file = NamedTempFile::new().unwrap();
        let test_content =
            b"Large benchmark test content for parallel processing. ".repeat(2000000); // ~100MB
        temp_file.write_all(&test_content).unwrap();
        temp_file.flush().unwrap();

        // Benchmark the function
        let start = std::time::Instant::now();
        let hash = hash_blake3_file(temp_file.path()).unwrap();
        let duration = start.elapsed();

        println!("BLAKE3 Large file hash: {}", hash);
        println!("BLAKE3 Large file time: {:?}", duration);
        println!("Large file size: {} bytes", test_content.len());
        println!(
            "BLAKE3 Large file speed: {:.2} MB/s",
            (test_content.len() as f64 / 1024.0 / 1024.0) / duration.as_secs_f64()
        );
        assert_eq!(
            hash,
            "6cd27b8098295afc42527bcee267ce39e757345f9f82c20b66efc75ffe4c1631"
        );
    }

    #[test]
    fn test_transform_github_url_blob() {
        // Test GitHub blob URL transformation
        let blob_url =
            "https://github.com/dathere/qsv/blob/master/resources/test/boston311-100.csv";
        let expected =
            "https://raw.githubusercontent.com/dathere/qsv/master/resources/test/boston311-100.csv";
        assert_eq!(transform_github_url(blob_url), expected);
    }

    #[test]
    fn test_transform_github_url_blob_with_branch() {
        // Test with a different branch name
        let blob_url = "https://github.com/user/repo/blob/develop/path/to/file.csv";
        let expected = "https://raw.githubusercontent.com/user/repo/develop/path/to/file.csv";
        assert_eq!(transform_github_url(blob_url), expected);
    }

    #[test]
    fn test_transform_github_url_blob_http() {
        // Test HTTP (not HTTPS) URL transformation
        let blob_url = "http://github.com/user/repo/blob/main/file.csv";
        let expected = "https://raw.githubusercontent.com/user/repo/main/file.csv";
        assert_eq!(transform_github_url(blob_url), expected);
    }

    #[test]
    fn test_transform_github_url_raw() {
        // Test that raw URLs are not modified
        let raw_url =
            "https://raw.githubusercontent.com/dathere/qsv/master/resources/test/boston311-100.csv";
        assert_eq!(transform_github_url(raw_url), raw_url);
    }

    #[test]
    fn test_transform_github_url_non_github() {
        // Test that non-GitHub URLs are not modified
        let other_url = "https://example.com/data/file.csv";
        assert_eq!(transform_github_url(other_url), other_url);
    }

    #[test]
    fn test_transform_github_url_no_blob() {
        // Test that GitHub URLs without /blob/ are not modified
        let repo_url = "https://github.com/dathere/qsv";
        assert_eq!(transform_github_url(repo_url), repo_url);
    }

    #[test]
    fn test_transform_github_url_with_query_params() {
        // Test that query parameters are stripped (they don't apply to raw.githubusercontent.com)
        let blob_url = "https://github.com/user/repo/blob/main/file.csv?ref=v1.0";
        let expected = "https://raw.githubusercontent.com/user/repo/main/file.csv";
        assert_eq!(transform_github_url(blob_url), expected);
    }

    #[test]
    fn test_transform_github_url_with_fragment() {
        // Test that fragments (line highlighting) are stripped
        let blob_url = "https://github.com/user/repo/blob/main/file.csv#L10-L20";
        let expected = "https://raw.githubusercontent.com/user/repo/main/file.csv";
        assert_eq!(transform_github_url(blob_url), expected);
    }

    #[test]
    fn test_transform_github_url_with_query_and_fragment() {
        // Test that both query parameters and fragments are stripped
        let blob_url = "https://github.com/user/repo/blob/main/file.csv?ref=v1.0#L10";
        let expected = "https://raw.githubusercontent.com/user/repo/main/file.csv";
        assert_eq!(transform_github_url(blob_url), expected);
    }
}
