static USAGE: &str = r#"
Find the difference between two CSVs with ludicrous speed.

NOTE: diff does not support stdin. A file path is required for both arguments.
      Further, PRIMARY KEY VALUES MUST BE UNIQUE WITHIN EACH CSV.

      To check if a CSV has unique primary key values, use `qsv extdedup`
      with the same key columns using the `--select` option:

         $ qsv extdedup --select keycol data.csv --no-output

      The duplicate count will be printed to stderr.

Examples:

# Find the difference between two CSVs
qsv diff left.csv right.csv

# Find the difference between two CSVs when the right CSV has no headers
qsv diff left.csv --no-headers-right right-noheaders.csv

# Find the difference between two CSVs when the left CSV uses a tab delimiter
qsv diff --delimiter-left '\t' left.csv right-tab.tsv

# Find the difference between two CSVs when the left CSV uses a semicolon delimiter
qsv diff --delimiter-left ';' left.csv right-semicolon.csv

# Find the difference between two CSVs and write output with tab delimiter to a file
qsv diff -o diff-tab.tsv --delimiter-output '\t' left.csv right.csv

# Find the difference between two CSVs and write output with semicolon delimiter to a file
qsv diff -o diff-semicolon.csv --delimiter-output ';' left.csv right.csv

# Find the difference comparing records with the same values in the first two columns
qsv diff --key 0,1 left.csv right.csv

# Find the difference using first two columns as key and sort result by those columns
qsv diff -k 0,1 --sort-columns 0,1 left.csv right.csv

# Find the difference but replace equal field values with empty string (key fields still appear)
qsv diff --drop-equal-fields left.csv right.csv

# Find the difference but drop entire columns that have no differences anywhere (key columns still appear)
qsv diff --drop-equal-columns left.csv right.csv

# Combine both: keep only differing columns, and within them, blank out equal field values
qsv diff --drop-equal-columns --drop-equal-fields left.csv right.csv

# Find the difference but do not output headers in the result
qsv diff --no-headers-output left.csv right.csv

# Find the difference when both CSVs have no headers (generic headers _col_1, _col_2, etc. are used)
qsv diff --no-headers-left --no-headers-right left.csv right.csv

For more examples, see https://github.com/dathere/qsv/blob/master/tests/test_diff.rs
See also https://github.com/dathere/qsv/wiki/Indexing-Compression-Diff#diff

Usage:
    qsv diff [options] [<input-left>] [<input-right>]
    qsv diff --help

diff options:
    --no-headers-left           When set, the first row will be considered as part of
                                the left CSV to diff. (When not set, the
                                first row is the header row and will be skipped during
                                the diff. It will always appear in the output.)
    --no-headers-right          When set, the first row will be considered as part of
                                the right CSV to diff. (When not set, the
                                first row is the header row and will be skipped during
                                the diff. It will always appear in the output.)
    --no-headers-output         When set, the diff result won't have a header row in
                                its output. If not set and both CSVs have no headers,
                                headers in the result will be: _col_1,_col_2, etc.
    --delimiter-left <arg>      The field delimiter for reading CSV data on the left.
                                Must be a single character. (default: ,)
    --delimiter-right <arg>     The field delimiter for reading CSV data on the right.
                                Must be a single character. (default: ,)
    --delimiter-output <arg>    The field delimiter for writing the CSV diff result.
                                Must be a single character. (default: ,)
    -k, --key <arg...>          The column indices that uniquely identify a record
                                as a comma separated list of 0-based indices, e.g. 0,1,2
                                or column names, e.g. name,age.
                                Note that when selecting columns by name, only the
                                left CSV's headers are used to match the column names
                                and it is assumed that the right CSV has the same
                                selected column names in the same order as the left CSV.
                                (default: 0)
    --sort-columns <arg...>     The column indices by which the diff result should be
                                sorted as a comma separated list of indices, e.g. 0,1,2
                                or column names, e.g. name,age.
                                Records in the diff result that are marked as "modified"
                                ("delete" and "add" records that have the same key,
                                but have different content) will always be kept together
                                in the sorted diff result and so won't be sorted
                                independently from each other.
                                Note that when selecting columns by name, only the
                                left CSV's headers are used to match the column names
                                and it is assumed that the right CSV has the same
                                selected column names in the same order as the left CSV.
    --drop-equal-fields         Drop values of equal fields in modified rows of the CSV
                                diff result (and replace them with the empty string).
                                Key field values will not be dropped.
    --drop-equal-columns        Drop entire columns from the diff result that have no
                                differences anywhere. A column is kept if it is a key
                                column, if it differs in any modified row, or if it has
                                a non-empty value in any added or deleted row. Otherwise
                                it is dropped. Can be combined with --drop-equal-fields.
    -j, --jobs <arg>            The number of jobs to run in parallel.
                                When not set, the number of jobs is set to the number
                                of CPUs detected.

Common options:
    -h, --help                  Display this message
    -o, --output <file>         Write output to <file> instead of stdout.
    -d, --delimiter <arg>       Set ALL delimiters to this character.
                                Overrides --delimiter-right, --delimiter-left
                                and --delimiter-output.
"#;

use std::io::{self, Write};

use csv::ByteRecord;
use csv_diff::{
    csv_diff::CsvByteDiffBuilder, csv_headers::Headers, diff_result::DiffByteRecords,
    diff_row::DiffByteRecord,
};
use serde::Deserialize;

use super::rename::rename_headers_all_generic;
use crate::{
    CliResult,
    clitypes::CliError,
    config::{Config, Delimiter},
    util,
};

#[derive(Deserialize)]
struct Args {
    arg_input_left:          Option<String>,
    arg_input_right:         Option<String>,
    flag_output:             Option<String>,
    flag_jobs:               Option<usize>,
    flag_no_headers_left:    bool,
    flag_no_headers_right:   bool,
    flag_no_headers_output:  bool,
    flag_delimiter_left:     Option<Delimiter>,
    flag_delimiter_right:    Option<Delimiter>,
    flag_delimiter_output:   Option<Delimiter>,
    flag_key:                Option<String>,
    flag_sort_columns:       Option<String>,
    flag_drop_equal_fields:  bool,
    flag_drop_equal_columns: bool,
    flag_delimiter:          Option<Delimiter>,
}

pub fn run(argv: &[&str]) -> CliResult<()> {
    let mut args: Args = util::get_args(USAGE, argv)?;

    if let Some(delim) = args.flag_delimiter {
        [
            args.flag_delimiter_left,
            args.flag_delimiter_right,
            args.flag_delimiter_output,
        ] = [Some(delim); 3];
    }

    let rconfig_left = Config::new(args.arg_input_left.as_ref())
        .delimiter(args.flag_delimiter_left)
        .no_headers(args.flag_no_headers_left);

    let rconfig_right = Config::new(args.arg_input_right.as_ref())
        .delimiter(args.flag_delimiter_right)
        .no_headers(args.flag_no_headers_right);

    if rconfig_left.is_stdin() || rconfig_right.is_stdin() {
        return fail_incorrectusage_clierror!(
            "diff does not support stdin. A file path is required for both arguments."
        );
    }

    let mut csv_rdr_left = rconfig_left.reader()?;
    let mut csv_rdr_right = rconfig_right.reader()?;

    let headers_left = csv_rdr_left.byte_headers()?;
    let headers_right = csv_rdr_right.byte_headers()?;

    let primary_key_cols: Vec<usize> = match args.flag_key {
        None => vec![0],
        Some(s) => parse_indices_or_colnames(&s, headers_left, headers_right, "key")?,
    };

    let sort_cols = args
        .flag_sort_columns
        .as_deref()
        .map(|s| parse_indices_or_colnames(s, headers_left, headers_right, "sort"))
        .transpose()?;

    let wtr = Config::new(args.flag_output.as_ref())
        .delimiter(args.flag_delimiter_output)
        .writer()?;

    util::njobs(args.flag_jobs);

    // ===== DIFF PROCESSING =====

    let csv_diff = CsvByteDiffBuilder::new()
        .primary_key_columns(primary_key_cols.clone())
        .build()
        .map_err(|e| CliError::Other(format!("Cannot instantiate diff: {e}")))?;

    let mut diff_byte_records = csv_diff
        .diff(csv_rdr_left.into(), csv_rdr_right.into())
        .try_to_diff_byte_records()?;

    match sort_cols {
        Some(sort_cols) => {
            diff_byte_records
                .sort_by_columns(sort_cols)
                .map_err(|e| CliError::Other(e.to_string()))?;
        },
        None => {
            diff_byte_records.sort_by_line();
        },
    }

    // When --drop-equal-columns is set, compute which columns to keep with a single
    // borrow-pass over the (already fully buffered) diff records. A column is kept if
    // it is a key column, if it differs in any modified row, or if it has a non-empty
    // value in any added or deleted row.
    let keep_cols: Option<Vec<usize>> = if args.flag_drop_equal_columns {
        let num_cols = diff_byte_records.num_columns().unwrap_or(0);
        let mut keep = vec![false; num_cols];
        for &k in &primary_key_cols {
            if k < num_cols {
                keep[k] = true;
            }
        }
        for dbr in diff_byte_records.iter() {
            match dbr {
                DiffByteRecord::Modify { field_indices, .. } => {
                    for &fi in field_indices {
                        if fi < num_cols {
                            keep[fi] = true;
                        }
                    }
                },
                DiffByteRecord::Add(rec) | DiffByteRecord::Delete(rec) => {
                    for (i, field) in rec.byte_record().iter().enumerate() {
                        if i < num_cols && !field.is_empty() {
                            keep[i] = true;
                        }
                    }
                },
            }
        }
        Some(
            keep.iter()
                .enumerate()
                .filter_map(|(i, &k)| k.then_some(i))
                .collect(),
        )
    } else {
        None
    };

    let mut csv_diff_writer = CsvDiffWriter::new(
        wtr,
        args.flag_no_headers_output,
        args.flag_drop_equal_fields,
        primary_key_cols,
        keep_cols,
    );
    Ok(csv_diff_writer.write_diff_byte_records(diff_byte_records)?)
}

trait StrExt {
    fn col_names_to_indices(
        &self,
        col_names_split_by: char,
        headers: &ByteRecord,
        msg_left_or_right: &str,
    ) -> Result<Vec<usize>, CliError>;
}

impl StrExt for str {
    fn col_names_to_indices(
        &self,
        col_names_split_by: char,
        headers: &ByteRecord,
        msg_left_or_right: &str,
    ) -> Result<Vec<usize>, CliError> {
        self.split(col_names_split_by)
            .map(|col_name| {
                headers
                    .iter()
                    .position(|h| h == col_name.as_bytes())
                    .ok_or_else(|| {
                        CliError::Other(format!(
                            "Column name '{col_name}' not found on {msg_left_or_right} CSV"
                        ))
                    })
            })
            .collect::<Result<Vec<usize>, _>>()
    }
}

/// Parse a comma-separated list of either 0-based column indices or column names.
/// When column names are given, both CSVs must resolve them to the same indices.
/// `kind` is used in error messages (e.g., "key" or "sort").
fn parse_indices_or_colnames(
    s: &str,
    headers_left: &ByteRecord,
    headers_right: &ByteRecord,
    kind: &str,
) -> Result<Vec<usize>, CliError> {
    if s.chars().all(|c| c.is_ascii_digit() || c == ',') {
        return s
            .split(',')
            .map(str::parse::<usize>)
            .collect::<Result<Vec<_>, _>>()
            .map_err(|err| {
                CliError::IncorrectUsage(format!("Invalid {kind} column index list '{s}': {err}"))
            });
    }

    let left_indices = s.col_names_to_indices(',', headers_left, "left")?;
    let right_indices = s.col_names_to_indices(',', headers_right, "right")?;

    if left_indices != right_indices {
        return fail_incorrectusage_clierror!(
            "Column names on left and right CSVs do not match.\nUse `qsv select` to reorder the \
             columns on the right CSV to match the order of the left CSV.\nThe {kind} column \
             indices on the left CSV are in index locations:\n{left_indices:?}\nand on the right \
             CSV are:\n{right_indices:?}",
        );
    }
    Ok(left_indices)
}

struct CsvDiffWriter<W: Write> {
    csv_writer:        csv::Writer<W>,
    no_headers:        bool,
    drop_equal_fields: bool,
    key_fields:        Vec<usize>,
    // When Some, only these (sorted) column indices are written to the output.
    // None means all columns are written (the default).
    keep_cols:         Option<Vec<usize>>,
}

impl<W: Write> CsvDiffWriter<W> {
    fn new(
        csv_writer: csv::Writer<W>,
        no_headers: bool,
        drop_equal_fields: bool,
        key_fields: impl IntoIterator<Item = usize>,
        keep_cols: Option<Vec<usize>>,
    ) -> Self {
        Self {
            csv_writer,
            no_headers,
            drop_equal_fields,
            key_fields: key_fields.into_iter().collect(),
            keep_cols,
        }
    }

    /// Project a full-width header `ByteRecord` down to `keep_cols` (or clone it
    /// unchanged when no columns are being dropped), then write it with the
    /// `diffresult` prefix column.
    fn write_projected_header(&mut self, header: &ByteRecord) -> csv::Result<()> {
        match &self.keep_cols {
            Some(keep) => {
                let mut projected = ByteRecord::new();
                for &c in keep {
                    projected.push_field(&header[c]);
                }
                projected.write_diffresult_header(&mut self.csv_writer)
            },
            None => header.write_diffresult_header(&mut self.csv_writer),
        }
    }

    /// Write a full-width row (prefix sign at index 0, then one entry per column),
    /// projecting it down to `keep_cols` when columns are being dropped.
    fn write_projected_record(&mut self, full_row: &[&[u8]]) -> csv::Result<()> {
        match &self.keep_cols {
            Some(keep) => {
                let mut projected: Vec<&[u8]> = Vec::with_capacity(keep.len() + 1);
                projected.push(full_row[0]);
                for &c in keep {
                    projected.push(full_row[c + 1]);
                }
                self.csv_writer.write_record(projected)
            },
            None => self.csv_writer.write_record(full_row),
        }
    }

    fn write_headers(&mut self, headers: &Headers, num_columns: Option<&usize>) -> csv::Result<()> {
        match (headers.headers_left(), headers.headers_right()) {
            (Some(lbh), Some(rbh)) => {
                // currently, `diff` can only handle two CSVs that have the same
                // headers ordering, so in this case we can either choose the left
                // or right headers, because both are the same
                debug_assert_eq!(
                    lbh, rbh,
                    "csv_diff invariant: left/right headers must match"
                );
                if !self.no_headers {
                    let lbh = lbh.clone();
                    self.write_projected_header(&lbh)?;
                }
            },
            (Some(bh), None) | (None, Some(bh)) => {
                if !self.no_headers {
                    let bh = bh.clone();
                    self.write_projected_header(&bh)?;
                }
            },
            (None, None) => {
                if let (Some(&num_cols), false) = (num_columns.filter(|&&c| c > 0), self.no_headers)
                {
                    let headers_generic = rename_headers_all_generic(num_cols);
                    let mut new_rdr = csv::Reader::from_reader(headers_generic.as_bytes());
                    let new_headers = new_rdr.byte_headers()?.clone();
                    self.write_projected_header(&new_headers)?;
                }
            },
        }

        Ok(())
    }

    fn write_diff_byte_records(&mut self, diff_byte_records: DiffByteRecords) -> io::Result<()> {
        self.write_headers(
            diff_byte_records.headers(),
            diff_byte_records.num_columns().as_ref(),
        )?;
        for dbr in diff_byte_records {
            self.write_diff_byte_record(&dbr)?;
        }
        self.csv_writer.flush()?;
        Ok(())
    }

    #[inline]
    fn write_diff_byte_record(&mut self, diff_byte_record: &DiffByteRecord) -> csv::Result<()> {
        let add_sign: &[u8] = &b"+"[..];
        let remove_sign: &[u8] = &b"-"[..];

        match diff_byte_record {
            DiffByteRecord::Add(add) => {
                let mut vec = vec![add_sign];
                vec.extend(add.byte_record());
                self.write_projected_record(&vec)
            },
            DiffByteRecord::Modify {
                delete,
                add,
                field_indices,
            } => {
                let vec_del = if self.drop_equal_fields {
                    self.fill_modified_and_drop_equal_fields(
                        remove_sign,
                        delete.byte_record(),
                        field_indices.as_slice(),
                    )
                } else {
                    let mut tmp = vec![remove_sign];
                    tmp.extend(delete.byte_record());
                    tmp
                };

                self.write_projected_record(&vec_del)?;

                let vec_add = if self.drop_equal_fields {
                    self.fill_modified_and_drop_equal_fields(
                        add_sign,
                        add.byte_record(),
                        field_indices.as_slice(),
                    )
                } else {
                    let mut tmp = vec![add_sign];
                    tmp.extend(add.byte_record());
                    tmp
                };

                self.write_projected_record(&vec_add)
            },
            DiffByteRecord::Delete(del) => {
                let mut vec = vec![remove_sign];
                vec.extend(del.byte_record());
                self.write_projected_record(&vec)
            },
        }
    }

    fn fill_modified_and_drop_equal_fields<'a>(
        &self,
        prefix: &'a [u8],
        byte_record: &'a ByteRecord,
        modified_field_indices: &[usize],
    ) -> Vec<&'a [u8]> {
        let mut vec_to_fill = {
            // We start out with all fields set to an empty byte slice
            // (which end up as our equal fields).
            let mut tmp = vec![&b""[..]; byte_record.len() + 1 /* + 1, because we need to store our additional prefix*/];
            tmp.as_mut_slice()[0] = prefix;
            tmp
        };
        // key field values and modified field values should appear in the output
        for &field_idx in self.key_fields.iter().chain(modified_field_indices) {
            // + 1 here, because of the prefix value (see above)
            vec_to_fill[field_idx + 1] = &byte_record[field_idx];
        }

        vec_to_fill
    }
}

trait WriteDiffResultHeader {
    fn write_diffresult_header<W: Write>(&self, csv_writer: &mut csv::Writer<W>)
    -> csv::Result<()>;
}

impl WriteDiffResultHeader for csv::ByteRecord {
    fn write_diffresult_header<W: Write>(
        &self,
        csv_writer: &mut csv::Writer<W>,
    ) -> csv::Result<()> {
        if !self.is_empty() {
            let mut new_header = vec![&b"diffresult"[..]];
            new_header.extend(self);
            csv_writer.write_record(new_header)?;
        }
        Ok(())
    }
}
