use crate::FwReaderSchema;
use crate::byte_record::ByteRecord;
use crate::byte_record_iterator::ByteRecordIterator;
use crate::fw_error::Result;
use crate::fw_reader_options::FwReaderOptions;
use crate::fw_reader_state::{FwReaderState, ReadOutcome};
#[cfg(feature = "encoding_rs")]
use crate::string_record::StringRecord;
#[cfg(feature = "encoding_rs")]
use crate::string_record_iterator::StringRecordIterator;
use std::io::Read;

/// A fixed-width format reader.
///
/// Reads records from a source implementing [`Read`]. Construct one via
/// [`FwReader::options`]:
///
/// ```
/// use fwf::{FwReader, FwReaderField, FwReaderSchema, RecordSeparator};
///
/// let schema = FwReaderSchema::builder(7)
///     .field(FwReaderField::builder(0..5).build())
///     .field(FwReaderField::builder(5..7).build())
///     .separator(RecordSeparator::Lf)
///     .build()
///     .unwrap();
///
/// let data = b"Alice30\nBob  25\n";
/// let mut reader = FwReader::options(schema).from_reader(data.as_slice());
///
/// while let Some(record) = reader.read_byte_record().unwrap() {
///     let _name = &record[0];
/// }
/// assert_eq!(reader.record_number(), 2);
/// ```
#[derive(Debug)]
pub struct FwReader<R> {
    reader: R,
    state: FwReaderState,
}

impl FwReader<()> {
    /// Creates a new [`FwReaderOptions`] for the given schema.
    #[must_use]
    #[inline]
    pub fn options(schema: FwReaderSchema) -> FwReaderOptions {
        FwReaderOptions::new(schema)
    }
}

impl<R> FwReader<R> {
    /// Creates a new reader. Called by [`FwReaderOptions`].
    pub(crate) fn new(
        reader: R,
        schema: FwReaderSchema,
        #[cfg(feature = "encoding_rs")] encoding: &'static encoding_rs::Encoding,
    ) -> Self {
        let state = FwReaderState::new(
            schema,
            #[cfg(feature = "encoding_rs")]
            encoding,
        );
        Self { reader, state }
    }

    /// Swaps the schema driving this reader and returns the new reader paired
    /// with the previous schema.
    ///
    /// The underlying reader's position is preserved, and the previous
    /// schema's internal buffers are returned along with it so it can be
    /// reused without reallocation. Use this to alternate between schemas
    /// when reading multi-record-shape formats.
    #[must_use]
    pub fn into_reader_for_schema(self, schema: FwReaderSchema) -> (FwReader<R>, FwReaderSchema) {
        let (new_state, old_schema) = self.state.for_schema(schema);
        let new_reader = Self {
            reader: self.reader,
            state: new_state,
        };
        (new_reader, old_schema)
    }
}

impl<R: Read> FwReader<R> {
    /// Returns the number of records successfully read so far.
    #[must_use]
    #[inline]
    pub fn record_number(&self) -> u64 {
        self.state.record_number()
    }

    /// Reads the next record as a [`ByteRecord`] borrowing from the internal buffer.
    ///
    /// Returns `Ok(None)` at the end of the source.
    ///
    /// # Errors
    ///
    /// Returns [`crate::FwfError::IncompleteRecord`] if the source ends with a partial
    /// record (fewer than `width` bytes remaining).
    ///
    /// Returns [`crate::FwfError::IncompleteSeparator`] if the source ends between the
    /// record data and the expected separator.
    ///
    /// Returns [`crate::FwfError::InvalidSeparator`] if the bytes read as a separator
    /// do not match the expected separator.
    ///
    /// Returns [`crate::FwfError::Io`] if the underlying reader produces an I/O error.
    pub fn read_byte_record(&mut self) -> Result<Option<ByteRecord<'_>>> {
        if !self.fill_record_buffer()? {
            return Ok(None);
        }
        Ok(Some(self.state.byte_record()))
    }

    /// Reads the next record and returns it as an owned `ByteRecord<'static>`,
    /// packing field data into a single contiguous buffer.
    pub(crate) fn read_owned_byte_record(&mut self) -> Result<Option<ByteRecord<'static>>> {
        if !self.fill_record_buffer()? {
            return Ok(None);
        }
        Ok(Some(self.state.owned_byte_record()))
    }

    /// Reads the next record as a [`StringRecord`] borrowing from the internal buffer.
    ///
    /// Returns `Ok(None)` at the end of the source.
    ///
    /// # Errors
    ///
    /// Returns [`crate::FwfError::Encoding`] if a field cannot be decoded with the
    /// configured encoding.
    ///
    /// Returns [`crate::FwfError::IncompleteRecord`] if the source ends with a partial
    /// record (fewer than `width` bytes remaining).
    ///
    /// Returns [`crate::FwfError::IncompleteSeparator`] if the source ends between the
    /// record data and the expected separator.
    ///
    /// Returns [`crate::FwfError::InvalidSeparator`] if the bytes read as a separator
    /// do not match the expected separator.
    ///
    /// Returns [`crate::FwfError::Io`] if the underlying reader produces an I/O error.
    #[cfg(feature = "encoding_rs")]
    pub fn read_string_record(&mut self) -> Result<Option<StringRecord<'_>>> {
        if !self.fill_record_buffer()? {
            return Ok(None);
        }
        self.state.string_record().map(Some)
    }

    /// Reads the next record and returns it as an owned `StringRecord<'static>`,
    /// packing decoded field text into a single contiguous `String`.
    #[cfg(feature = "encoding_rs")]
    pub(crate) fn read_owned_string_record(&mut self) -> Result<Option<StringRecord<'static>>> {
        if !self.fill_record_buffer()? {
            return Ok(None);
        }
        self.state.owned_string_record().map(Some)
    }

    /// Reads the next record into the internal buffer. Returns `false` at EOF.
    fn fill_record_buffer(&mut self) -> Result<bool> {
        if self.state.is_complete() {
            return Ok(false);
        }

        let outcome = read_fill(&mut self.reader, self.state.buffer_mut())?;
        if !self.state.handle_record_read(outcome)? {
            return Ok(false);
        }

        if self.state.has_separator() {
            let outcome = read_fill(&mut self.reader, self.state.separator_buffer_mut())?;
            self.state.handle_separator_read(outcome)?;
        }

        self.state.fill_field_ranges();
        Ok(true)
    }

    /// Returns an iterator over records as [`ByteRecord<'static>`].
    #[must_use]
    #[inline]
    pub fn byte_records(self) -> ByteRecordIterator<R> {
        ByteRecordIterator::new(self)
    }

    /// Returns an iterator over records as [`StringRecord<'static>`].
    #[cfg(feature = "encoding_rs")]
    #[must_use]
    #[inline]
    pub fn string_records(self) -> StringRecordIterator<R> {
        StringRecordIterator::new(self)
    }
}

/// Attempts to fill `buf` from a sync reader, returning the outcome.
pub(crate) fn read_fill(
    reader: &mut impl Read,
    buf: &mut [u8],
) -> std::result::Result<ReadOutcome, std::io::Error> {
    let mut bytes_read = 0;
    while bytes_read < buf.len() {
        match reader.read(&mut buf[bytes_read..]) {
            Ok(0) => break,
            Ok(n) => bytes_read += n,
            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
            Err(e) => return Err(e),
        }
    }

    if bytes_read == 0 {
        Ok(ReadOutcome::Eof)
    } else if bytes_read < buf.len() {
        Ok(ReadOutcome::Partial(bytes_read))
    } else {
        Ok(ReadOutcome::Filled)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::Alignment;
    use crate::FwfError;
    use crate::fw_reader_field::FwReaderField;
    use crate::record_separator::RecordSeparator;

    fn name_age_reader(data: &[u8], separator: RecordSeparator) -> FwReader<&[u8]> {
        let schema = FwReaderSchema::builder(7)
            .field(FwReaderField::builder(0..5).build())
            .field(FwReaderField::builder(5..7).build())
            .separator(separator)
            .build()
            .unwrap();
        FwReader::options(schema).from_reader(data)
    }

    // -- No separator --

    #[test]
    fn read_no_separator() {
        let data = b"Alice30Bob  25";
        let mut reader = name_age_reader(data.as_slice(), RecordSeparator::None);

        let rec = reader.read_byte_record().unwrap().unwrap();
        assert_eq!(&rec[0], b"Alice");
        assert_eq!(&rec[1], b"30");

        let rec = reader.read_byte_record().unwrap().unwrap();
        assert_eq!(&rec[0], b"Bob");
        assert_eq!(&rec[1], b"25");

        assert!(reader.read_byte_record().unwrap().is_none());
        assert_eq!(reader.record_number(), 2);
    }

    // -- LF separator --

    #[test]
    fn read_lf_separator() {
        let data = b"Alice30\nBob  25\n";
        let mut reader = name_age_reader(data.as_slice(), RecordSeparator::Lf);

        let rec = reader.read_byte_record().unwrap().unwrap();
        assert_eq!(&rec[0], b"Alice");
        assert_eq!(&rec[1], b"30");

        let rec = reader.read_byte_record().unwrap().unwrap();
        assert_eq!(&rec[0], b"Bob");
        assert_eq!(&rec[1], b"25");

        assert!(reader.read_byte_record().unwrap().is_none());
    }

    #[test]
    fn read_lf_separator_missing_on_last_record() {
        let data = b"Alice30\nBob  25";
        let mut reader = name_age_reader(data.as_slice(), RecordSeparator::Lf);

        reader.read_byte_record().unwrap().unwrap();
        let rec = reader.read_byte_record().unwrap().unwrap();
        assert_eq!(&rec[0], b"Bob");
        assert_eq!(&rec[1], b"25");

        assert!(reader.read_byte_record().unwrap().is_none());
    }

    // -- CRLF separator --

    #[test]
    fn read_crlf_separator() {
        let data = b"Alice30\r\nBob  25\r\n";
        let mut reader = name_age_reader(data.as_slice(), RecordSeparator::CrLf);

        let rec = reader.read_byte_record().unwrap().unwrap();
        assert_eq!(&rec[0], b"Alice");

        let rec = reader.read_byte_record().unwrap().unwrap();
        assert_eq!(&rec[0], b"Bob");

        assert!(reader.read_byte_record().unwrap().is_none());
    }

    // -- CR separator --

    #[test]
    fn read_cr_separator() {
        let data = b"Alice30\rBob  25\r";
        let mut reader = name_age_reader(data.as_slice(), RecordSeparator::Cr);

        reader.read_byte_record().unwrap().unwrap();
        reader.read_byte_record().unwrap().unwrap();
        assert!(reader.read_byte_record().unwrap().is_none());
    }

    // -- Custom separator --

    #[test]
    fn read_custom_separator() {
        let data = b"Alice30||Bob  25||";
        let mut reader = name_age_reader(data.as_slice(), RecordSeparator::Custom(b"||".to_vec()));

        let rec = reader.read_byte_record().unwrap().unwrap();
        assert_eq!(&rec[0], b"Alice");

        let rec = reader.read_byte_record().unwrap().unwrap();
        assert_eq!(&rec[0], b"Bob");

        assert!(reader.read_byte_record().unwrap().is_none());
    }

    // -- Padding / alignment --

    #[test]
    fn read_strips_left_aligned_padding() {
        let data = b"Alice     30";
        let schema = FwReaderSchema::builder(12)
            .field(FwReaderField::builder(0..10).build())
            .field(FwReaderField::builder(10..12).build())
            .build()
            .unwrap();
        let mut reader = FwReader::options(schema).from_reader(data.as_slice());

        let rec = reader.read_byte_record().unwrap().unwrap();
        assert_eq!(&rec[0], b"Alice");
        assert_eq!(&rec[1], b"30");
    }

    #[test]
    fn read_strips_right_aligned_padding() {
        let data = b"     Alice30";
        let schema = FwReaderSchema::builder(12)
            .field(
                FwReaderField::builder(0..10)
                    .alignment(Alignment::Right)
                    .build(),
            )
            .field(FwReaderField::builder(10..12).build())
            .build()
            .unwrap();
        let mut reader = FwReader::options(schema).from_reader(data.as_slice());

        let rec = reader.read_byte_record().unwrap().unwrap();
        assert_eq!(&rec[0], b"Alice");
        assert_eq!(&rec[1], b"30");
    }

    #[test]
    fn read_strips_custom_padding() {
        let data = b"00042";
        let schema = FwReaderSchema::builder(5)
            .field(
                FwReaderField::builder(0..5)
                    .padding(b'0')
                    .alignment(Alignment::Right)
                    .build(),
            )
            .build()
            .unwrap();
        let mut reader = FwReader::options(schema).from_reader(data.as_slice());

        let rec = reader.read_byte_record().unwrap().unwrap();
        assert_eq!(&rec[0], b"42");
    }

    // -- Overlapping fields --

    #[test]
    fn read_overlapping_fields() {
        let data = b"ABCDEFGHIJ";
        let schema = FwReaderSchema::builder(10)
            .field(FwReaderField::builder(0..5).build())
            .field(FwReaderField::builder(3..8).build())
            .build()
            .unwrap();
        let mut reader = FwReader::options(schema).from_reader(data.as_slice());

        let rec = reader.read_byte_record().unwrap().unwrap();
        assert_eq!(&rec[0], b"ABCDE");
        assert_eq!(&rec[1], b"DEFGH");
    }

    // -- Error cases --

    #[test]
    fn read_incomplete_record() {
        let data = b"ABC";
        let schema = FwReaderSchema::builder(5)
            .field(FwReaderField::builder(0..5).build())
            .build()
            .unwrap();
        let mut reader = FwReader::options(schema).from_reader(data.as_slice());

        let err = reader.read_byte_record().unwrap_err();
        assert!(matches!(
            err,
            FwfError::IncompleteRecord {
                bytes_read: 3,
                expected: 5
            }
        ));
    }

    #[test]
    fn read_incomplete_separator() {
        let data = b"Alice30\r";
        let mut reader = name_age_reader(data.as_slice(), RecordSeparator::CrLf);

        let err = reader.read_byte_record().unwrap_err();
        assert!(matches!(
            err,
            FwfError::IncompleteSeparator {
                bytes_read: 1,
                expected: 2
            }
        ));
    }

    #[test]
    fn read_invalid_separator() {
        let data = b"Alice30XXBob  25\n";
        let mut reader = name_age_reader(data.as_slice(), RecordSeparator::CrLf);

        let err = reader.read_byte_record().unwrap_err();
        assert!(matches!(err, FwfError::InvalidSeparator));
    }

    #[test]
    fn read_empty_input() {
        let data = b"";
        let schema = FwReaderSchema::builder(5)
            .field(FwReaderField::builder(0..5).build())
            .build()
            .unwrap();
        let mut reader = FwReader::options(schema).from_reader(data.as_slice());

        assert!(reader.read_byte_record().unwrap().is_none());
        assert_eq!(reader.record_number(), 0);
    }

    // -- Iterator --

    #[test]
    fn byte_records_iterator() {
        let data = b"Alice30\nBob  25\n";
        let reader = name_age_reader(data.as_slice(), RecordSeparator::Lf);

        let records: Vec<_> = reader
            .byte_records()
            .collect::<std::result::Result<Vec<_>, _>>()
            .unwrap();
        assert_eq!(records.len(), 2);
        assert_eq!(&records[0][0], b"Alice");
        assert_eq!(&records[1][0], b"Bob");
    }

    // -- record_number tracking --

    #[test]
    fn record_number_increments() {
        let data = b"AABBBCC";
        let schema = FwReaderSchema::builder(7)
            .field(FwReaderField::builder(0..2).build())
            .build()
            .unwrap();
        let mut reader = FwReader::options(schema).from_reader(data.as_slice());

        assert_eq!(reader.record_number(), 0);
        reader.read_byte_record().unwrap();
        assert_eq!(reader.record_number(), 1);
    }
}

#[cfg(test)]
#[cfg(feature = "encoding_rs")]
mod encoding_tests {
    use super::*;
    use crate::FwfError;
    use crate::fw_reader_field::FwReaderField;
    use crate::record_separator::RecordSeparator;

    fn name_age_reader(data: &[u8], separator: RecordSeparator) -> FwReader<&[u8]> {
        let schema = FwReaderSchema::builder(7)
            .field(FwReaderField::builder(0..5).build())
            .field(FwReaderField::builder(5..7).build())
            .separator(separator)
            .build()
            .unwrap();
        FwReader::options(schema).from_reader(data)
    }

    #[test]
    fn read_string_record_ascii() {
        let data = b"Alice30\nBob  25\n";
        let mut reader = name_age_reader(data.as_slice(), RecordSeparator::Lf);

        let rec = reader.read_string_record().unwrap().unwrap();
        assert_eq!(&rec[0], "Alice");
        assert_eq!(&rec[1], "30");

        let rec = reader.read_string_record().unwrap().unwrap();
        assert_eq!(&rec[0], "Bob");
        assert_eq!(&rec[1], "25");

        assert!(reader.read_string_record().unwrap().is_none());
    }

    #[test]
    fn read_string_record_utf8() {
        let data = "café!42".as_bytes();
        let schema = FwReaderSchema::builder(8)
            .field(FwReaderField::builder(0..6).build())
            .field(FwReaderField::builder(6..8).build())
            .build()
            .unwrap();
        let mut reader = FwReader::options(schema).from_reader(data);

        let rec = reader.read_string_record().unwrap().unwrap();
        assert_eq!(&rec[0], "café!");
        assert_eq!(&rec[1], "42");
    }

    #[test]
    fn read_string_record_utf8_split_multibyte_returns_encoding_error() {
        let data = "café".as_bytes();
        let schema = FwReaderSchema::builder(5)
            .field(FwReaderField::builder(0..4).build())
            .build()
            .unwrap();
        let mut reader = FwReader::options(schema).from_reader(data);

        let err = reader.read_string_record().unwrap_err();
        assert!(matches!(err, FwfError::Encoding { field_index: 0 }));
    }

    #[test]
    fn read_string_record_utf8_split_emoji_returns_encoding_error() {
        let data = "A😊B".as_bytes();
        let schema = FwReaderSchema::builder(6)
            .field(FwReaderField::builder(0..3).build())
            .field(FwReaderField::builder(3..6).build())
            .build()
            .unwrap();
        let mut reader = FwReader::options(schema).from_reader(data);

        let err = reader.read_string_record().unwrap_err();
        assert!(matches!(err, FwfError::Encoding { field_index: 0 }));
    }

    #[test]
    fn read_string_record_utf8_clean_boundary() {
        let data = "Aé B".as_bytes();
        let schema = FwReaderSchema::builder(5)
            .field(FwReaderField::builder(0..3).build())
            .field(FwReaderField::builder(3..5).build())
            .build()
            .unwrap();
        let mut reader = FwReader::options(schema).from_reader(data);

        let rec = reader.read_string_record().unwrap().unwrap();
        assert_eq!(&rec[0], "Aé");
        assert_eq!(&rec[1], " B");
    }

    #[test]
    fn string_records_iterator() {
        let data = b"Alice30\nBob  25\n";
        let reader = name_age_reader(data.as_slice(), RecordSeparator::Lf);

        let records: Vec<_> = reader
            .string_records()
            .collect::<std::result::Result<Vec<_>, _>>()
            .unwrap();
        assert_eq!(records.len(), 2);
        assert_eq!(&records[0][0], "Alice");
        assert_eq!(&records[0][1], "30");
        assert_eq!(&records[1][0], "Bob");
        assert_eq!(&records[1][1], "25");
    }

    #[test]
    fn string_record_into_owned() {
        let data = b"Hello42";
        let schema = FwReaderSchema::builder(7)
            .field(FwReaderField::builder(0..5).build())
            .field(FwReaderField::builder(5..7).build())
            .build()
            .unwrap();
        let mut reader = FwReader::options(schema).from_reader(data.as_slice());

        let rec = reader.read_string_record().unwrap().unwrap();
        let owned = rec.into_owned();
        assert_eq!(&owned[0], "Hello");
        assert_eq!(&owned[1], "42");
    }
}
