use std::{
    env, fs,
    io::{self, Read},
    path::{Path, PathBuf},
    sync::{
        Arc, OnceLock,
        atomic::{AtomicBool, Ordering},
    },
};

use csv_nose::{SampleSize, Sniffer};
use log::{debug, info, warn};
use serde::de::{Deserialize, Deserializer, Error};

use crate::{
    CliResult,
    index::Indexed,
    select::{SelectColumns, Selection},
    util,
};

// rdr default is 8k in csv crate, we're making it 128k
pub const DEFAULT_RDR_BUFFER_CAPACITY: usize = 128 * (1 << 10);
// previous wtr default in xsv is 32k, we're making it 512k
pub const DEFAULT_WTR_BUFFER_CAPACITY: usize = 512 * (1 << 10);

// number of rows for csv-nose to sample
const DEFAULT_SNIFFER_SAMPLE: usize = 100;

// file size at which we warn user that a large file has not been indexed
const NO_INDEX_WARNING_FILESIZE: u64 = 100 * (1 << 20); // 100MB

// so we don't have to keep checking if the index has been created
static AUTO_INDEXED: AtomicBool = AtomicBool::new(false);

pub static SPONSOR_MESSAGE: &str = r"sponsored by datHere - Data Infrastructure Engineering (https://qsv.datHere.com)
Need a UI & more advanced data-wrangling? Upgrade to qsv pro (https://qsvpro.datHere.com)
";

pub static TEMP_FILE_DIR: OnceLock<PathBuf> = OnceLock::new();

#[cfg(feature = "polars")]
pub static POLARS_FLOAT_PRECISION: OnceLock<Option<usize>> = OnceLock::new();

// Variants are constructed by `get_special_format` but only meaningfully matched
// when the `polars` feature is enabled (via `util::convert_special_format`),
// so non-polars builds see them as never read.
//
// Exception: `CompressedZip` is both *detected* and *handled* in all builds. It
// is handled by `util::extract_zip_to_temp` (always compiled — it needs only the
// non-optional `zip` crate), and `Config::new` preserves `CompressedZip` even in
// non-polars builds (mapping only the other, polars-only variants to `Unknown`).
#[allow(dead_code)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SpecialFormat {
    Avro,
    Parquet,
    Ipc,
    Json,  // expects JSON Array
    Jsonl, // expects JSON Lines
    CompressedCsv,
    CompressedTsv,
    CompressedSsv,
    CompressedZip,
    Unknown,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Delimiter(pub u8);

/// Delimiter represents values that can be passed from the command line that
/// can be used as a field delimiter in CSV data.
///
/// Its purpose is to ensure that the Unicode character given decodes to a
/// valid ASCII character as required by the CSV parser.
impl Delimiter {
    pub const fn as_byte(self) -> u8 {
        self.0
    }

    pub fn decode_delimiter(s: &str) -> Result<Delimiter, String> {
        if s == r"\t" {
            return Ok(Delimiter(b'\t'));
        }

        if s.len() != 1 {
            return fail_format!("Could not convert '{s}' to a single ASCII character.");
        }

        let c = s.chars().next().unwrap();
        if c.is_ascii() {
            Ok(Delimiter(c as u8))
        } else {
            fail_format!("Could not convert '{c}' to ASCII delimiter.")
        }
    }
}

impl<'de> Deserialize<'de> for Delimiter {
    fn deserialize<D: Deserializer<'de>>(d: D) -> Result<Delimiter, D::Error> {
        let s = String::deserialize(d)?;
        match Delimiter::decode_delimiter(&s) {
            Ok(delim) => Ok(delim),
            Err(msg) => Err(D::Error::custom(msg)),
        }
    }
}

#[derive(Clone, Debug)]
pub struct Config {
    pub path:              Option<PathBuf>, // None implies <stdin>
    idx_path:              Option<PathBuf>,
    select_columns:        Option<SelectColumns>,
    delimiter:             u8,
    // The detected special input format (Parquet/Avro/Arrow/JSON/JSONL, compressed
    // CSV, or zip). Conversion to a delimited temp file is deferred to the read
    // path (see `prepared_for_read`), so a Config used only for writing never
    // converts its (output) path. `Unknown` for ordinary inputs, stdin & writers.
    special_format:        SpecialFormat,
    // Lazily-resolved, cached (shared across clones) converted-input (temp path,
    // delimiter) for `special_format` inputs. Populated on first read.
    read_input:            Arc<OnceLock<Result<(PathBuf, u8), String>>>,
    pub no_headers:        bool,
    pub flexible:          bool,
    terminator:            csv::Terminator,
    pub quote:             u8,
    quote_style:           csv::QuoteStyle,
    double_quote:          bool,
    escape:                Option<u8>,
    quoting:               bool,
    pub preamble_rows:     u64,
    trim:                  csv::Trim,
    pub autoindex_size:    u64,
    prefer_dmy:            bool,
    pub comment:           Option<u8>,
    snappy:                bool, // flag to enable snappy compression/decompression
    pub read_buffer:       u32,
    pub write_buffer:      u32,
    pub skip_format_check: bool,
    pub format_error:      Option<String>,
}

// Empty trait as an alias for Seek and Read that avoids auto trait errors
pub trait SeekRead: io::Seek + io::Read {}
impl<T: io::Seek + io::Read> SeekRead for T {}

/// Parse the named env var as `T`, falling back to `default` if it is unset or invalid.
/// Logs a warning if the env var is set but cannot be parsed.
fn parse_env_or_warn<T: std::str::FromStr + std::fmt::Display>(name: &str, default: T) -> T {
    match env::var(name) {
        Ok(s) => s.parse().unwrap_or_else(|_| {
            warn!("invalid {name} value {s:?}; using default {default}");
            default
        }),
        Err(_) => default,
    }
}

impl Config {
    /// Creates a new `Config` instance with default settings and optional file path.
    ///
    /// # Arguments
    ///
    /// * `path` - An optional reference to a `String` representing the file path.
    ///
    /// # Returns
    ///
    /// A new `Config` instance.
    ///
    /// # Details
    ///
    /// This function initializes a `Config` with the following behavior:
    /// - Uses env var `QSV_DEFAULT_DELIMITER` for default delimiter, or ',' if not set
    /// - Determines delimiter and Snappy compression based on file extension.
    /// - Supports sniffing delimiter and preamble rows if `QSV_SNIFF_DELIMITER` or
    ///   `QSV_SNIFF_PREAMBLE` is set.
    /// - Sets comment character from `QSV_COMMENT_CHAR` environment variable.
    /// - Sets headers behavior based on `QSV_NO_HEADERS` environment variable.
    /// - Configures various other settings from environment variables.
    ///
    /// # Environment Variables
    ///
    /// - `QSV_DEFAULT_DELIMITER`: Sets the default delimiter.
    /// - `QSV_SNIFF_DELIMITER` or `QSV_SNIFF_PREAMBLE`: Enables sniffing of delimiter and preamble
    ///   rows.
    /// - `QSV_COMMENT_CHAR`: Sets the comment character.
    /// - `QSV_NO_HEADERS`: Determines if the file has headers.
    /// - `QSV_AUTOINDEX_SIZE`: Sets the auto-index size.
    /// - `QSV_PREFER_DMY`: Sets date format preference.
    /// - `QSV_RDR_BUFFER_CAPACITY`: Sets read buffer capacity.
    /// - `QSV_WTR_BUFFER_CAPACITY`: Sets write buffer capacity.
    /// - `QSV_SKIP_FORMAT_CHECK`: Set to skip file extension checking.
    pub fn new(path: Option<&String>) -> Config {
        // Resolve a `dc:<name>` cache reference (the `get` command's disk cache)
        // to a concrete file path up front, so the rest of Config::new treats it
        // like an ordinary local file (delimiter detection, indexing, etc.).
        // Config::new is infallible, so a resolution failure is surfaced later
        // via `format_error` rather than panicking.
        #[cfg(feature = "get")]
        let mut dc_format_error: Option<String> = None;
        #[cfg(feature = "get")]
        let dc_resolved: Option<String> = match path {
            Some(s) if s.starts_with("dc:") => {
                match crate::diskcache::resolve_dc_path(&s["dc:".len()..]) {
                    Ok(p) => Some(p.to_string_lossy().to_string()),
                    Err(e) => {
                        dc_format_error = Some(e.to_string());
                        None
                    },
                }
            },
            _ => None,
        };
        #[cfg(feature = "get")]
        let path: Option<&String> = dc_resolved.as_ref().or(path);

        let default_delim = match env::var("QSV_DEFAULT_DELIMITER") {
            Ok(delim) => match Delimiter::decode_delimiter(&delim) {
                Ok(d) => d.as_byte(),
                Err(e) => {
                    warn!("invalid QSV_DEFAULT_DELIMITER {delim:?} ({e}); using ','");
                    b','
                },
            },
            _ => b',',
        };
        let mut sniff = util::get_envvar_flag("QSV_SNIFF_DELIMITER")
            || util::get_envvar_flag("QSV_SNIFF_PREAMBLE");
        let mut skip_format_check = true;
        let mut format_error = None;
        let (path, mut delim, snappy, special_format) = match path {
            None => (None, default_delim, false, SpecialFormat::Unknown),
            // WIP: support remote files; currently only http(s) is supported
            // Some(ref s) if s.starts_with("http") && Url::parse(s).is_ok() => {
            //     let mut snappy = false;
            //     let delim = if s.ends_with(".csv.sz") {
            //         snappy = true;
            //         b','
            //     } else if s.ends_with(".tsv.sz") || s.ends_with(".tab.sz") {
            //         snappy = true;
            //         b'\t'
            //     } else {
            //         default_delim
            //     };
            //     // download the file to a temporary location
            //     util::download_file()
            //     (Some(PathBuf::from(s)), delim, snappy)
            // },
            Some(s) if s == "-" => (None, default_delim, false, SpecialFormat::Unknown),
            Some(s) => {
                let path = PathBuf::from(s);

                // if QSV_SKIP_FORMAT_CHECK is set or path is a temp file, we skip format check
                let temp_dir = crate::config::TEMP_FILE_DIR.get_or_init(|| {
                    tempfile::TempDir::new().map_or_else(
                        |e| {
                            warn!(
                                "failed to create temp dir: {e}; falling back to system temp dir"
                            );
                            env::temp_dir()
                        },
                        tempfile::TempDir::keep,
                    )
                });
                skip_format_check = sniff
                    || util::get_envvar_flag("QSV_SKIP_FORMAT_CHECK")
                    || path.starts_with(temp_dir);

                // Detect special formats. The actual conversion to a delimited temp
                // file is DEFERRED to the read path (see `prepared_for_read`), so a
                // Config used only for writing never converts its (output) path.
                // `.zip` is detected even without polars (it needs only the `zip`
                // crate); the other special formats require polars to convert and so
                // stay `Unknown` otherwise.
                #[cfg(feature = "polars")]
                let special_format = get_special_format(&path);
                #[cfg(not(feature = "polars"))]
                let special_format = if get_special_format(&path) == SpecialFormat::CompressedZip {
                    SpecialFormat::CompressedZip
                } else {
                    SpecialFormat::Unknown
                };

                // Delimiter/snappy come from the path's own extension. For special
                // formats this is the write/fallback delimiter; the read delimiter
                // is re-derived from the converted temp in `prepared_for_read`.
                let (file_extension, delim, snappy) = get_delim_by_extension(&path, default_delim);

                if special_format == SpecialFormat::Unknown {
                    // Only ordinary inputs are subject to the extension check.
                    format_error = if skip_format_check {
                        None
                    } else {
                        match file_extension.as_str() {
                            "csv" | "tsv" | "tab" | "ssv" => None,
                            ext => Some(format!(
                                "{} is using an unsupported file format: {ext}. Set \
                                 QSV_SKIP_FORMAT_CHECK to skip input format checking.",
                                path.display()
                            )),
                        }
                    };
                } else {
                    // Don't sniff a binary/compressed special-format file; the
                    // converted temp is what actually gets read.
                    sniff = false;
                }
                (Some(path), delim, snappy, special_format)
            },
        };
        let comment: Option<u8> = env::var("QSV_COMMENT_CHAR")
            .ok()
            .and_then(|s| s.as_bytes().first().copied());
        let no_headers = util::get_envvar_flag("QSV_NO_HEADERS");
        let mut preamble = 0_u64;
        if let (true, Some(sniff_path_buf)) = (sniff, path.as_ref()) {
            if let Some(sniff_path) = sniff_path_buf.to_str() {
                match Sniffer::new()
                    .sample_size(SampleSize::Records(DEFAULT_SNIFFER_SAMPLE))
                    .sniff_path(sniff_path)
                {
                    Ok(metadata) => {
                        delim = metadata.dialect.delimiter;
                        preamble = metadata.dialect.header.num_preamble_rows as u64;
                        info!(
                            "sniffed delimiter {} and {preamble} preamble rows",
                            delim as char
                        );
                    },
                    // we only warn, as we don't want to stop processing the file
                    // if sniffing doesn't work
                    Err(e) => warn!("sniff error: {e}"),
                }
            } else {
                warn!(
                    "skipping delimiter sniff: path {} is not valid UTF-8",
                    sniff_path_buf.display()
                );
            }
        }

        // A failed `dc:` resolution takes precedence so its (more actionable)
        // error is what surfaces at reader-construction time.
        #[cfg(feature = "get")]
        let format_error = dc_format_error.or(format_error);

        Config {
            path,
            idx_path: None,
            select_columns: None,
            delimiter: delim,
            special_format,
            read_input: Arc::new(OnceLock::new()),
            no_headers,
            flexible: false,
            terminator: csv::Terminator::Any(b'\n'),
            quote: b'"',
            quote_style: csv::QuoteStyle::Necessary,
            double_quote: true,
            escape: None,
            quoting: true,
            preamble_rows: preamble,
            trim: csv::Trim::None,
            autoindex_size: parse_env_or_warn("QSV_AUTOINDEX_SIZE", 0_u64),
            prefer_dmy: util::get_envvar_flag("QSV_PREFER_DMY"),
            comment,
            snappy,
            read_buffer: parse_env_or_warn(
                "QSV_RDR_BUFFER_CAPACITY",
                DEFAULT_RDR_BUFFER_CAPACITY as u32,
            ),
            write_buffer: parse_env_or_warn(
                "QSV_WTR_BUFFER_CAPACITY",
                DEFAULT_WTR_BUFFER_CAPACITY as u32,
            ),
            format_error,
            skip_format_check,
        }
    }

    pub const fn delimiter(mut self, d: Option<Delimiter>) -> Config {
        if let Some(d) = d {
            self.delimiter = d.as_byte();
        }
        self
    }

    pub fn get_delimiter(&self) -> u8 {
        // For a special-format input the effective delimiter is the converted
        // temp's (e.g. a `.tsv` entry inside a `.zip`), not the outer path's
        // default. Resolve (cached) so callers see the real delimiter regardless
        // of whether they've read yet; fall back to the configured delimiter if
        // conversion fails.
        if self.special_format != SpecialFormat::Unknown
            && let Ok((_, delim)) = self.resolve_converted()
        {
            return delim;
        }
        self.delimiter
    }

    pub const fn comment(mut self, c: Option<u8>) -> Config {
        self.comment = c;
        self
    }

    pub const fn get_dmy_preference(&self) -> bool {
        self.prefer_dmy
    }

    /// Explicitly set `no_headers`, unconditionally overriding env var.
    /// Use this when a command knows the input has (or lacks) headers
    /// regardless of user configuration (e.g. internally-generated CSVs).
    pub const fn no_headers(mut self, yes: bool) -> Config {
        self.no_headers = yes;
        self
    }

    /// Apply the `--no-headers` CLI flag without overriding `QSV_NO_HEADERS` env var.
    /// When the flag is `false` (not passed), the env var value is preserved.
    /// When the flag is `true` (explicitly passed), it sets `no_headers = true`.
    /// Also respects `QSV_TOGGLE_HEADERS` to flip the flag value.
    pub fn no_headers_flag(mut self, mut yes: bool) -> Config {
        if env::var("QSV_TOGGLE_HEADERS").unwrap_or_else(|_| "0".to_owned()) == "1" {
            yes = !yes;
        }
        self.no_headers = self.no_headers || yes;
        self
    }

    pub const fn flexible(mut self, yes: bool) -> Config {
        self.flexible = yes;
        self
    }

    pub const fn skip_format_check(mut self, yes: bool) -> Config {
        self.skip_format_check = yes;
        self
    }

    #[cfg(any(feature = "feature_capable", feature = "lite"))]
    pub const fn crlf(mut self, yes: bool) -> Config {
        if yes {
            self.terminator = csv::Terminator::CRLF;
        } else {
            self.terminator = csv::Terminator::Any(b'\n');
        }
        self
    }

    #[cfg(any(feature = "feature_capable", feature = "lite"))]
    pub const fn terminator(mut self, term: csv::Terminator) -> Config {
        self.terminator = term;
        self
    }

    pub const fn quote(mut self, quote: u8) -> Config {
        self.quote = quote;
        self
    }

    pub const fn quote_style(mut self, style: csv::QuoteStyle) -> Config {
        self.quote_style = style;
        self
    }

    pub const fn double_quote(mut self, yes: bool) -> Config {
        self.double_quote = yes;
        self
    }

    pub const fn escape(mut self, escape: Option<u8>) -> Config {
        self.escape = escape;
        self
    }

    pub const fn quoting(mut self, yes: bool) -> Config {
        self.quoting = yes;
        self
    }

    pub const fn trim(mut self, trim_type: csv::Trim) -> Config {
        self.trim = trim_type;
        self
    }

    pub fn set_read_buffer(mut self, buffer: usize) -> Config {
        self.read_buffer = u32::try_from(buffer).unwrap_or_else(|_| {
            warn!(
                "read buffer {buffer} exceeds u32::MAX; using default \
                 {DEFAULT_RDR_BUFFER_CAPACITY}"
            );
            DEFAULT_RDR_BUFFER_CAPACITY as u32
        });
        self
    }

    pub fn set_write_buffer(mut self, buffer: usize) -> Config {
        self.write_buffer = u32::try_from(buffer).unwrap_or_else(|_| {
            warn!(
                "write buffer {buffer} exceeds u32::MAX; using default \
                 {DEFAULT_WTR_BUFFER_CAPACITY}"
            );
            DEFAULT_WTR_BUFFER_CAPACITY as u32
        });
        self
    }

    #[allow(clippy::missing_const_for_fn)]
    pub fn select(mut self, sel_cols: SelectColumns) -> Config {
        self.select_columns = Some(sel_cols);
        self
    }

    pub const fn is_stdin(&self) -> bool {
        self.path.is_none()
    }

    #[cfg(feature = "polars")]
    pub const fn is_snappy(&self) -> bool {
        self.snappy
    }

    #[inline]
    /// Returns a `Selection` based on the config's `select_columns` & the first record of the CSV.
    ///
    /// # Arguments
    ///
    /// * `first_record` - A reference to the first `ByteRecord` of the CSV.
    ///
    /// # Returns
    ///
    /// * `Result<Selection, String>` - A `Selection` if successful, otherwise, an error msg
    ///
    /// # Errors
    ///
    /// This function will return an error if:
    /// * The `Config` has no `SelectColumns` (i.e., `Config::select` was not called).
    pub fn selection(&self, first_record: &csv::ByteRecord) -> Result<Selection, String> {
        match self.select_columns {
            None => fail!("Config has no 'SelectColumns'. Did you call Config::select?"),
            Some(ref sel) => sel.selection(first_record, !self.no_headers),
        }
    }

    /// Writes the headers from a CSV reader to a CSV writer.
    ///
    /// This function reads the headers from the given CSV reader and writes them to the CSV writer,
    /// but only if the `no_headers` flag is not set. If the headers are empty, nothing is written.
    ///
    /// # Arguments
    ///
    /// * `r` - A mutable reference to a CSV reader.
    /// * `w` - A mutable reference to a CSV writer.
    ///
    /// # Returns
    ///
    /// Returns a `csv::Result<()>` which is `Ok(())` if the operation was successful,
    /// or an error if there was a problem reading or writing.
    pub fn write_headers<R: io::Read, W: io::Write>(
        &self,
        r: &mut csv::Reader<R>,
        w: &mut csv::Writer<W>,
    ) -> csv::Result<()> {
        if !self.no_headers {
            let r = r.byte_headers()?;
            if !r.is_empty() {
                w.write_record(r)?;
            }
        }
        Ok(())
    }

    pub fn writer(&self) -> io::Result<csv::Writer<Box<dyn io::Write + 'static>>> {
        Ok(self.from_writer(self.io_writer()?))
    }

    /// Lazily convert a `special_format` input to a delimited temp file, returning
    /// the temp path and its delimiter. The result is cached and shared across
    /// clones, so a Config that is read more than once converts only once. With an
    /// explicit `QSV_SKIP_FORMAT_CHECK`, a conversion failure falls back to reading
    /// the original bytes as-is rather than erroring.
    fn resolve_converted(&self) -> io::Result<(PathBuf, u8)> {
        // safety: only called when special_format != Unknown, which implies a path
        let src = self
            .path
            .as_ref()
            .expect("special-format Config must have a path");
        let cached = self.read_input.get_or_init(|| {
            match util::convert_special_format(src, self.special_format, self.delimiter) {
                Ok(temp) => {
                    let (_, delim, _) = get_delim_by_extension(&temp, self.delimiter);
                    Ok((temp, delim))
                },
                Err(e) => Err(format!("Failed to convert special format: {e}")),
            }
        });
        match cached {
            Ok((p, d)) => Ok((p.clone(), *d)),
            Err(e) => {
                // A detected special format that fails to convert is a hard error
                // for ALL formats — reading the raw (binary) bytes as delimited text
                // silently produces garbage with a success exit code. The only escape
                // hatch is an explicit QSV_SKIP_FORMAT_CHECK, which means "I know what
                // I'm doing, read the original bytes as-is".
                if util::get_envvar_flag("QSV_SKIP_FORMAT_CHECK") {
                    Ok((src.clone(), self.delimiter))
                } else {
                    Err(io::Error::new(io::ErrorKind::InvalidInput, e.clone()))
                }
            },
        }
    }

    /// The path that will actually be read. For a `special_format` input this
    /// lazily converts to (and returns the path of) the delimited temp file, so
    /// callers that need the *decompressed* size — e.g. `util::mem_file_check`
    /// memory guards — see the real data rather than the compressed source. For
    /// ordinary inputs and stdin, returns `path` unchanged. The conversion is
    /// cached, so the subsequent reader reuses the same temp.
    pub fn resolved_path(&self) -> CliResult<Option<PathBuf>> {
        if self.special_format == SpecialFormat::Unknown {
            return Ok(self.path.clone());
        }
        Ok(Some(self.resolve_converted()?.0))
    }

    /// Returns a Config ready to *read* this input. For a `special_format` input,
    /// it is a clone whose `path` points at the lazily-converted delimited temp
    /// (with that temp's delimiter) and whose `special_format` is `Unknown`, so the
    /// read methods treat it as an ordinary delimited file (no re-entry). Only
    /// called from the read entry points when `special_format != Unknown`.
    pub(crate) fn prepared_for_read(&self) -> io::Result<Config> {
        if self.special_format == SpecialFormat::Unknown {
            return Ok(self.clone());
        }
        let (temp, delim) = self.resolve_converted()?;
        let mut c = self.clone();
        c.path = Some(temp);
        c.delimiter = delim;
        c.special_format = SpecialFormat::Unknown;
        Ok(c)
    }

    pub fn reader(&self) -> io::Result<csv::Reader<Box<dyn io::Read + Send + 'static>>> {
        if self.special_format != SpecialFormat::Unknown {
            return self.prepared_for_read()?.reader();
        }
        if !self.skip_format_check && self.format_error.is_some() {
            Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                self.format_error.clone().unwrap(),
            ))
        } else {
            Ok(self.from_reader(self.io_reader()?))
        }
    }

    pub fn reader_file(&self) -> io::Result<csv::Reader<fs::File>> {
        if self.special_format != SpecialFormat::Unknown {
            return self.prepared_for_read()?.reader_file();
        }
        match self.path {
            None => Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "Cannot use <stdin> here",
            )),
            Some(ref p) => {
                if !self.skip_format_check && self.format_error.is_some() {
                    Err(io::Error::new(
                        io::ErrorKind::InvalidInput,
                        self.format_error.clone().unwrap(),
                    ))
                } else {
                    fs::File::open(p).map(|f| self.from_reader(f))
                }
            },
        }
    }

    pub fn reader_file_stdin(&self) -> io::Result<csv::Reader<Box<dyn SeekRead + 'static>>> {
        if self.special_format != SpecialFormat::Unknown {
            return self.prepared_for_read()?.reader_file_stdin();
        }
        Ok(match self.path {
            None => {
                // Create a buffer in memory for stdin
                let mut buffer: Vec<u8> = Vec::new();
                let stdin = io::stdin();
                stdin.lock().read_to_end(&mut buffer)?;
                self.from_reader(Box::new(io::Cursor::new(buffer)))
            },
            Some(ref p) => {
                if !self.skip_format_check && self.format_error.is_some() {
                    return Err(io::Error::new(
                        io::ErrorKind::InvalidInput,
                        self.format_error.clone().unwrap(),
                    ));
                }
                self.from_reader(Box::new(fs::File::open(p)?))
            },
        })
    }

    /// Automatically creates an index file for the CSV file.
    ///
    /// This function attempts to create an index file for the CSV file specified in `self.path`.
    /// It's designed to fail silently if any step of the process encounters an error, as it's
    /// intended to be a convenience function.
    ///
    /// # Behavior
    ///
    /// - If the file is Snappy-compressed, the function returns immediately w/o creating an index.
    /// - If `self.path` is `None`, the function returns without action.
    /// - The function creates an index file using `util::idx_path()` to determine index file path.
    /// - It uses `csv_index::RandomAccessSimple::create()` to generate the index.
    /// - If index creation is successful, it sets the `AUTO_INDEXED` atomic flag to `true`.
    ///
    /// # Errors
    ///
    /// While this function doesn't return any errors, it logs debug messages for both successful
    /// and failed index creation attempts.
    fn autoindex_file(&self) {
        if self.snappy {
            return;
        }

        let Some(path_buf) = &self.path else { return };

        let pidx = util::idx_path(Path::new(path_buf));
        let Ok(idxfile) = fs::File::create(pidx) else {
            return;
        };
        let Ok(mut rdr) = self.reader_file() else {
            return;
        };
        let mut wtr = io::BufWriter::with_capacity(DEFAULT_WTR_BUFFER_CAPACITY, idxfile);
        match csv_index::RandomAccessSimple::create(&mut rdr, &mut wtr) {
            Ok(()) => {
                let Ok(()) = io::Write::flush(&mut wtr) else {
                    return;
                };
                debug!("autoindex of {} successful.", path_buf.display());
                AUTO_INDEXED.store(true, Ordering::Relaxed);
            },
            Err(e) => debug!("autoindex of {} failed: {e}", path_buf.display()),
        }
    }

    /// Check if the index file exists and is newer than the CSV file.
    /// If so, return the CSV file handle and the index file handle. If not, return None.
    /// Unless the CSV's file size >= `QSV_AUTOINDEX_SIZE`, then we'll create an index
    /// automatically. Stale indices (CSV newer than index) are rebuilt automatically, but only
    /// on the `(Some(path), None)` branch that resolves the index path internally; the
    /// `auto_indexed` and explicit-`(path, idx_path)` branches skip the staleness recheck.
    pub fn index_files(&self) -> io::Result<Option<(csv::Reader<fs::File>, fs::File)>> {
        if self.special_format != SpecialFormat::Unknown {
            return self.prepared_for_read()?.index_files();
        }
        // Track the data file's mtime and the resolved index path *only* on the
        // path that may need a staleness recheck. For the auto_indexed and
        // explicit-(path, idx_path) branches, staleness is not re-checked, so
        // these stay at their default values.
        let mut data_modified = 0_u64;
        let data_fsize;
        let mut idx_path_work: Option<PathBuf> = None;

        // the auto_indexed flag is set when an index is created automatically with
        // autoindex_file(). We use this flag to avoid checking if the index exists every
        // time this function is called. If the index was already auto-indexed, we can just
        // use it & return immediately.
        let auto_indexed = AUTO_INDEXED.load(Ordering::Relaxed);

        let (csv_file, mut idx_file) = if auto_indexed {
            (
                fs::File::open(self.path.clone().unwrap())?,
                fs::File::open(util::idx_path(&self.path.clone().unwrap()))?,
            )
        } else {
            match (&self.path, &self.idx_path) {
                (&None, &None) => return Ok(None),
                (&None, &Some(_)) => {
                    return Err(io::Error::new(
                        io::ErrorKind::InvalidInput,
                        "Cannot use <stdin> with indexes",
                    ));
                },
                // When the caller supplies both paths explicitly, trust them and skip
                // the staleness recheck below (idx_path_work stays None).
                (Some(p), Some(ip)) => (fs::File::open(p)?, fs::File::open(ip)?),
                (Some(p), &None) => {
                    // We generally don't want to report an error here, since we're
                    // passively trying to find an index.

                    (data_modified, data_fsize) = util::file_metadata(&p.metadata()?);
                    let idx_path = util::idx_path(p);
                    let idx_file = match fs::File::open(&idx_path) {
                        Err(_) => {
                            // the index file doesn't exist
                            if self.snappy {
                                // cannot index snappy compressed files
                                return Ok(None);
                            } else if self.autoindex_size > 0 && data_fsize >= self.autoindex_size {
                                // if CSV file size >= QSV_AUTOINDEX_SIZE, and
                                // its not a snappy file, create an index automatically
                                self.autoindex_file();
                                fs::File::open(&idx_path)?
                            } else if data_fsize >= NO_INDEX_WARNING_FILESIZE {
                                // warn user that the CSV file is large and not indexed
                                use indicatif::HumanBytes;

                                warn!(
                                    "The {} CSV file is larger than the {} \
                                     NO_INDEX_WARNING_FILESIZE threshold. Consider creating an \
                                     index file as it will make qsv commands much faster.",
                                    HumanBytes(data_fsize),
                                    HumanBytes(NO_INDEX_WARNING_FILESIZE)
                                );
                                return Ok(None);
                            } else {
                                // CSV not greater than QSV_AUTOINDEX_SIZE, and not greater than
                                // NO_INDEX_WARNING_FILESIZE, so we don't create an index
                                return Ok(None);
                            }
                        },
                        Ok(f) => f,
                    };
                    idx_path_work = Some(idx_path);
                    (fs::File::open(p)?, idx_file)
                },
            }
        };
        // If the CSV data was last modified after the index file was last
        // modified, recreate the stale index automatically. Only checked when
        // we resolved the index path ourselves (idx_path_work is Some).
        if let Some(idx_path) = &idx_path_work {
            let (idx_modified, _) = util::file_metadata(&idx_file.metadata()?);
            if data_modified > idx_modified {
                info!("index stale... autoindexing...");
                self.autoindex_file();
                idx_file = fs::File::open(idx_path)?;
            }
        }

        let csv_rdr = self.from_reader(csv_file);
        Ok(Some((csv_rdr, idx_file)))
    }

    /// Check if the index file exists and is newer than the CSV file.
    /// If so, return the index file.
    /// If not, return None.
    /// Unless `QSV_AUTOINDEX_SIZE` is set, in which case, we'll recreate the
    /// stale index automatically
    #[inline]
    pub fn indexed(&self) -> CliResult<Option<Indexed<fs::File, fs::File>>> {
        match self.index_files()? {
            None => Ok(None),
            Some((r, i)) => Ok(Some(Indexed::open(r, i)?)),
        }
    }

    pub fn io_reader(&self) -> io::Result<Box<dyn io::Read + Send + 'static>> {
        if self.special_format != SpecialFormat::Unknown {
            return self.prepared_for_read()?.io_reader();
        }
        Ok(match self.path {
            None => Box::new(io::stdin()),
            Some(ref p) => match fs::File::open(p) {
                Ok(x) => {
                    if self.snappy {
                        // Validate that the file is actually a snappy-compressed file
                        // before attempting decompression. This prevents "corrupt input" errors
                        // when a plain CSV file is incorrectly detected as snappy.
                        match util::is_valid_snappy_file(p) {
                            Ok(true) => {
                                info!("decoding snappy-compressed file: {}", p.display());
                                Box::new(snap::read::FrameDecoder::new(x))
                            },
                            Ok(false) => {
                                warn!(
                                    "File {} has .sz extension but is not a valid Snappy file. \
                                     Reading as plain file.",
                                    p.display()
                                );
                                Box::new(x)
                            },
                            Err(e) => {
                                warn!(
                                    "Failed to validate Snappy file {}: {}. Reading as plain file.",
                                    p.display(),
                                    e
                                );
                                Box::new(x)
                            },
                        }
                    } else {
                        Box::new(x)
                    }
                },
                Err(err) => {
                    let msg = format!("failed to open {}: {}", p.display(), err);
                    return Err(io::Error::new(io::ErrorKind::NotFound, msg));
                },
            },
        })
    }

    #[allow(clippy::wrong_self_convention)]
    pub fn from_reader<R: Read>(&self, rdr: R) -> csv::Reader<R> {
        csv::ReaderBuilder::new()
            .flexible(self.flexible)
            .delimiter(self.delimiter)
            .has_headers(!self.no_headers)
            .quote(self.quote)
            .quoting(self.quoting)
            .escape(self.escape)
            .buffer_capacity(self.read_buffer as usize)
            .comment(self.comment)
            .trim(self.trim)
            .from_reader(rdr)
    }

    pub fn io_writer(&self) -> io::Result<Box<dyn io::Write + 'static>> {
        // `path` is the user-specified path (special-format conversion is deferred
        // to the read path and never rewrites it), so writers target it directly.
        Ok(match self.path {
            None => Box::new(io::stdout()),
            Some(ref p) => {
                if p == "sink" {
                    // sink is /dev/null
                    Box::new(io::sink())
                } else if self.snappy {
                    info!("writing snappy-compressed file: {}", p.display());
                    Box::new(snap::write::FrameEncoder::new(fs::File::create(p)?))
                } else {
                    Box::new(fs::File::create(p)?)
                }
            },
        })
    }

    #[allow(clippy::wrong_self_convention)]
    pub fn from_writer<W: io::Write>(&self, mut wtr: W) -> csv::Writer<W> {
        if util::get_envvar_flag("QSV_OUTPUT_BOM")
            && let Err(e) = wtr.write_all("\u{FEFF}".as_bytes())
        {
            // BOM is best-effort: a broken pipe here would otherwise abort the
            // whole process. Log and let the next real write surface the error.
            warn!("failed to write UTF-8 BOM: {e}");
        }

        csv::WriterBuilder::new()
            .flexible(self.flexible)
            .delimiter(self.delimiter)
            .terminator(self.terminator)
            .quote(self.quote)
            .quote_style(self.quote_style)
            .double_quote(self.double_quote)
            .escape(self.escape.unwrap_or(b'\\'))
            .buffer_capacity(self.write_buffer as usize)
            .from_writer(wtr)
    }
}

/// Checks if a file path has a Snappy compression extension (.sz).
///
/// # Arguments
///
/// * `path` - A reference to the `Path` of the file.
///
/// # Returns
///
/// `true` if the file has a `.sz` extension (case-insensitive), `false` otherwise.
///
/// # Details
///
/// This function uses Rust's `Path::extension()` method which properly handles
/// multiple extensions (e.g., `file.csv.sz` → `Some("sz")`). It performs
/// case-insensitive comparison for robustness.
#[inline]
pub fn is_snappy_extension(path: &Path) -> bool {
    path.extension()
        .is_some_and(|ext| ext.eq_ignore_ascii_case("sz"))
}

/// This function examines the file extension to determine:
/// 1. The appropriate delimiter (tab for .tsv/.tab, semicolon for .ssv, comma for .csv).
/// 2. Whether the file is Snappy-compressed (indicated by a .sz extension).
/// 3. For Snappy-compressed files, it checks the extension before .sz to determine the delimiter.
///
/// If the file extension doesn't match known types, it returns the default delimiter.
pub fn get_delim_by_extension(path: &Path, default_delim: u8) -> (String, u8, bool) {
    let snappy = is_snappy_extension(path);

    // Get the extension before .sz if it's a snappy file, otherwise get the normal extension
    let file_extension = if snappy {
        // For snappy files like file.csv.sz, we need to get "csv"
        // We can do this by getting the file stem, then checking its extension
        path.file_stem()
            .and_then(|stem| Path::new(stem).extension())
            .and_then(|ext| ext.to_str())
            .unwrap_or("")
            .to_ascii_lowercase()
    } else {
        path.extension()
            .and_then(|ext| ext.to_str())
            .unwrap_or("")
            .to_ascii_lowercase()
    };

    let delim = match file_extension.as_str() {
        "tsv" | "tab" => b'\t',
        "ssv" => b';',
        "csv" => b',',
        _ => default_delim,
    };

    (file_extension, delim, snappy)
}

/// Determines if a file is a Parquet, Arrow IPC, JSONL, or compressed CSV file.
///
/// # Arguments
///
/// * `path` - A reference to the `Path` of the file.
///
/// # Returns
///
/// A `SpecialFormat` enum value indicating the type of special format the file is.
pub fn get_special_format(path: &Path) -> SpecialFormat {
    if !path.exists() {
        return SpecialFormat::Unknown;
    }

    let extension = path.extension().unwrap_or_default();
    match extension
        .to_str()
        .unwrap_or_default()
        .to_ascii_lowercase()
        .as_str()
    {
        "avro" => SpecialFormat::Avro,
        "parquet" => SpecialFormat::Parquet,
        "ipc" | "arrow" => SpecialFormat::Ipc,
        "jsonl" | "ndjson" => SpecialFormat::Jsonl,
        "json" => SpecialFormat::Json,
        "gz" | "zst" | "zlib" => compressed_csv_format(path),
        // zip is detected at the outer-extension level (not via
        // `compressed_csv_format`), since the inner entry's name — and thus the
        // delimiter — is only knowable after opening the archive.
        "zip" => SpecialFormat::CompressedZip,
        _ => SpecialFormat::Unknown,
    }
}

/// For a path like `data.csv.gz`, classify the inner CSV-family extension
/// (`csv`, `tsv`/`tab`, or `ssv`) into a `SpecialFormat::Compressed*` variant.
/// Returns `Unknown` if the inner extension is missing or not a known CSV family.
fn compressed_csv_format(path: &Path) -> SpecialFormat {
    let inner_ext = path
        .file_stem()
        .and_then(|stem| Path::new(stem).extension())
        .and_then(|ext| ext.to_str())
        .map(str::to_ascii_lowercase);
    match inner_ext.as_deref() {
        Some("csv") => SpecialFormat::CompressedCsv,
        Some("tsv" | "tab") => SpecialFormat::CompressedTsv,
        Some("ssv") => SpecialFormat::CompressedSsv,
        _ => SpecialFormat::Unknown,
    }
}

#[cfg(test)]
mod tests {
    use std::path::PathBuf;

    use super::*;

    #[test]
    fn test_csv_extension() {
        let path = PathBuf::from("test.csv");
        let (ext, delim, snappy) = get_delim_by_extension(&path, b',');
        assert_eq!(ext, "csv");
        assert_eq!(delim, b',');
        assert!(!snappy);
    }

    #[test]
    fn test_tsv_extension() {
        let path = PathBuf::from("test.tsv");
        let (ext, delim, snappy) = get_delim_by_extension(&path, b',');
        assert_eq!(ext, "tsv");
        assert_eq!(delim, b'\t');
        assert!(!snappy);
    }

    #[test]
    fn test_ssv_extension() {
        let path = PathBuf::from("test.ssv");
        let (ext, delim, snappy) = get_delim_by_extension(&path, b',');
        assert_eq!(ext, "ssv");
        assert_eq!(delim, b';');
        assert!(!snappy);
    }

    #[test]
    fn test_snappy_csv_extension() {
        let path = PathBuf::from("test.csv.sz");
        let (ext, delim, snappy) = get_delim_by_extension(&path, b',');
        assert_eq!(ext, "csv");
        assert_eq!(delim, b',');
        assert!(snappy);
    }

    #[test]
    fn test_snappy_tsv_extension() {
        let path = PathBuf::from("test.tsv.sz");
        let (ext, delim, snappy) = get_delim_by_extension(&path, b',');
        assert_eq!(ext, "tsv");
        assert_eq!(delim, b'\t');
        assert!(snappy);
    }

    #[test]
    fn test_unknown_extension() {
        let path = PathBuf::from("test.unknown");
        let default_delim = b'|';
        let (ext, delim, snappy) = get_delim_by_extension(&path, default_delim);
        assert_eq!(ext, "unknown");
        assert_eq!(delim, default_delim);
        assert!(!snappy);
    }

    #[test]
    fn test_no_extension() {
        let path = PathBuf::from("test");
        let default_delim = b',';
        let (ext, delim, snappy) = get_delim_by_extension(&path, default_delim);
        assert_eq!(ext, "");
        assert_eq!(delim, default_delim);
        assert!(!snappy);
    }

    #[test]
    fn test_is_snappy_extension_lowercase() {
        assert!(is_snappy_extension(Path::new("file.csv.sz")));
        assert!(is_snappy_extension(Path::new("file.sz")));
        assert!(is_snappy_extension(Path::new("file.tsv.sz")));
    }

    #[test]
    fn test_is_snappy_extension_uppercase() {
        assert!(is_snappy_extension(Path::new("file.csv.SZ")));
        assert!(is_snappy_extension(Path::new("file.SZ")));
    }

    #[test]
    fn test_is_snappy_extension_mixed_case() {
        assert!(is_snappy_extension(Path::new("file.csv.Sz")));
        assert!(is_snappy_extension(Path::new("file.sZ")));
    }

    #[test]
    fn test_is_snappy_extension_not_snappy() {
        assert!(!is_snappy_extension(Path::new("file.csv")));
        assert!(!is_snappy_extension(Path::new("file.gz")));
        assert!(!is_snappy_extension(Path::new("file")));
        assert!(!is_snappy_extension(Path::new("file.sz.backup")));
        // Test that extensions ending with "sz" but not exactly "sz" don't trigger detection
        assert!(!is_snappy_extension(Path::new("file.esz")));
        assert!(!is_snappy_extension(Path::new("file.KYpPcb8esz")));
        assert!(!is_snappy_extension(Path::new("data.esz")));
    }

    #[test]
    fn test_snappy_ssv_extension() {
        let path = PathBuf::from("test.ssv.sz");
        let (ext, delim, snappy) = get_delim_by_extension(&path, b',');
        assert_eq!(ext, "ssv");
        assert_eq!(delim, b';');
        assert!(snappy);
    }

    #[test]
    fn test_snappy_case_insensitive() {
        let path = PathBuf::from("test.csv.SZ");
        let (ext, delim, snappy) = get_delim_by_extension(&path, b',');
        assert_eq!(ext, "csv");
        assert_eq!(delim, b',');
        assert!(snappy);

        let path = PathBuf::from("test.TSV.sz");
        let (ext, delim, snappy) = get_delim_by_extension(&path, b',');
        assert_eq!(ext, "tsv");
        assert_eq!(delim, b'\t');
        assert!(snappy);
    }
}
