use tokio::io::AsyncReadExt;

use crate::FwReaderSchema;
use crate::byte_record::ByteRecord;
use crate::fw_error::Result;
use crate::fw_reader_state::{FwReaderState, ReadOutcome};
#[cfg(feature = "encoding_rs")]
use crate::string_record::StringRecord;

/// An async fixed-width format reader, powered by tokio.
///
/// This is the async counterpart of [`FwReader`](crate::FwReader). Construct
/// one via [`FwReader::options`](crate::FwReader::options) using the
/// `from_tokio_reader` or `from_tokio_file` terminal methods.
///
/// ```ignore
/// let schema = FwReaderSchema::builder(7)
///     .field(FwReaderField::builder(0..5).build())
///     .separator(RecordSeparator::Lf)
///     .build()?;
///
/// let file = tokio::fs::File::open("data.fwf").await?;
/// let mut reader = FwReader::options(schema).from_tokio_file(file);
///
/// while let Some(record) = reader.read_byte_record().await? {
///     let _name = &record[0];
/// }
/// ```
#[derive(Debug)]
pub struct AsyncFwReader<R> {
    reader: R,
    state: FwReaderState,
}

impl<R> AsyncFwReader<R> {
    /// Creates a new async reader. Called by [`FwReaderOptions`](crate::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,
    ) -> (AsyncFwReader<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: tokio::io::AsyncRead + Unpin> AsyncFwReader<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 an error on I/O failure, incomplete records, or separator
    /// mismatches. See [`FwReader::read_byte_record`](crate::FwReader::read_byte_record)
    /// for the full list.
    pub async fn read_byte_record(&mut self) -> Result<Option<ByteRecord<'_>>> {
        if !self.fill_record_buffer().await? {
            return Ok(None);
        }
        Ok(Some(self.state.byte_record()))
    }

    /// Reads the next record and returns it as an owned `ByteRecord<'static>`.
    ///
    /// # Errors
    ///
    /// Same error conditions as [`read_byte_record`](Self::read_byte_record).
    pub async fn read_owned_byte_record(&mut self) -> Result<Option<ByteRecord<'static>>> {
        if !self.fill_record_buffer().await? {
            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
    ///
    /// Same error conditions as [`read_byte_record`](Self::read_byte_record),
    /// plus [`crate::FwfError::Encoding`](crate::FwfError::Encoding) on decode failure.
    #[cfg(feature = "encoding_rs")]
    pub async fn read_string_record(&mut self) -> Result<Option<StringRecord<'_>>> {
        if !self.fill_record_buffer().await? {
            return Ok(None);
        }
        self.state.string_record().map(Some)
    }

    /// Reads the next record and returns it as an owned `StringRecord<'static>`.
    ///
    /// # Errors
    ///
    /// Same error conditions as [`read_string_record`](Self::read_string_record).
    #[cfg(feature = "encoding_rs")]
    pub async fn read_owned_string_record(&mut self) -> Result<Option<StringRecord<'static>>> {
        if !self.fill_record_buffer().await? {
            return Ok(None);
        }
        self.state.owned_string_record().map(Some)
    }

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

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

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

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

/// Attempts to fill `buf` from an async reader, returning the outcome.
async fn async_read_fill(
    reader: &mut (impl tokio::io::AsyncRead + Unpin),
    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..]).await {
            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 crate::fw_reader_field::FwReaderField;
    use crate::record_separator::RecordSeparator;
    use crate::{FwReader, FwReaderSchema};

    #[tokio::test]
    async fn async_read_byte_records() {
        let data = b"Alice30\nBob  25\n";
        let schema = FwReaderSchema::builder(7)
            .field(FwReaderField::builder(0..5).build())
            .field(FwReaderField::builder(5..7).build())
            .separator(RecordSeparator::Lf)
            .build()
            .unwrap();
        let mut reader = FwReader::options(schema).from_tokio_reader(data.as_slice());

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

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

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

    #[tokio::test]
    async fn async_read_owned_byte_record() {
        let data = b"Alice30";
        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_tokio_reader(data.as_slice());

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

    #[tokio::test]
    async fn async_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_tokio_reader(data.as_slice());

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

#[cfg(test)]
#[cfg(feature = "encoding_rs")]
mod async_encoding_tests {
    use crate::fw_reader_field::FwReaderField;
    use crate::record_separator::RecordSeparator;
    use crate::{FwReader, FwReaderSchema};

    #[tokio::test]
    async fn async_read_string_record() {
        let data = b"Alice30\nBob  25\n";
        let schema = FwReaderSchema::builder(7)
            .field(FwReaderField::builder(0..5).build())
            .field(FwReaderField::builder(5..7).build())
            .separator(RecordSeparator::Lf)
            .build()
            .unwrap();
        let mut reader = FwReader::options(schema).from_tokio_reader(data.as_slice());

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

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

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