use tokio::io::AsyncWriteExt;

use crate::FwWriterSchema;
use crate::fw_error::Result;
use crate::fw_writer_state::FwWriterState;

/// An async fixed-width format writer, powered by tokio.
///
/// This is the async counterpart of [`FwWriter`](crate::FwWriter). Construct
/// one via [`FwWriter::options`](crate::FwWriter::options) using the
/// `from_tokio_writer` or `from_tokio_file` terminal methods.
///
/// ```ignore
/// let schema = FwWriterSchema::builder()
///     .field(FwWriterField::builder(5).build())
///     .separator(RecordSeparator::Lf)
///     .build();
///
/// let file = tokio::fs::File::create("data.fwf").await?;
/// let mut writer = FwWriter::options(schema).from_tokio_file(file);
///
/// writer.write_byte_record(&[b"Alice"]).await?;
/// writer.flush().await?;
/// ```
#[derive(Debug)]
pub struct AsyncFwWriter<W> {
    writer: W,
    state: FwWriterState,
}

impl<W> AsyncFwWriter<W> {
    /// Creates a new async writer. Called by [`FwWriterOptions`](crate::FwWriterOptions).
    #[must_use]
    pub(crate) fn new(writer: W, state: FwWriterState) -> Self {
        Self { writer, state }
    }

    /// Swaps the schema driving this writer and returns the new writer paired
    /// with the previous schema.
    ///
    /// The underlying destination is preserved, and the previous schema's
    /// internal buffer is returned along with it so it can be reused without
    /// reallocation. Use this to alternate between schemas when emitting
    /// multi-record-shape formats.
    #[must_use]
    pub fn into_writer_for_schema(
        self,
        schema: FwWriterSchema,
    ) -> (AsyncFwWriter<W>, FwWriterSchema) {
        let (new_state, old_schema) = self.state.for_schema(schema);
        let new_writer = Self {
            writer: self.writer,
            state: new_state,
        };
        (new_writer, old_schema)
    }
}

impl<W: tokio::io::AsyncWrite + Unpin> AsyncFwWriter<W> {
    /// Returns the number of records successfully written so far.
    #[must_use]
    #[inline]
    pub fn record_number(&self) -> u64 {
        self.state.record_number()
    }

    /// Flushes the underlying writer.
    ///
    /// # Errors
    ///
    /// Returns [`crate::FwfError::Io`] if the flush fails.
    pub async fn flush(&mut self) -> Result<()> {
        self.writer.flush().await?;
        Ok(())
    }

    /// Consumes the writer and returns the underlying writer.
    #[must_use]
    #[inline]
    pub fn into_inner(self) -> W {
        self.writer
    }

    /// Writes a single record.
    ///
    /// The `values` slice must contain one entry per field in the schema.
    /// To produce a padding-only column, pass an empty `b""` value.
    ///
    /// # Errors
    ///
    /// Returns [`crate::FwfError::ValueCountMismatch`] if `values.len()` does not
    /// match the number of fields.
    ///
    /// Returns [`crate::FwfError::ValueTooLong`] if any value exceeds its field width,
    /// and the effective truncation policy is [`crate::TruncationPolicy::Error`].
    ///
    /// Returns [`crate::FwfError::Io`] on write failure.
    pub async fn write_byte_record(&mut self, values: &[&[u8]]) -> Result<()> {
        self.state.prepare_byte_record(values)?;
        self.writer.write_all(self.state.record_buffer()).await?;
        Ok(())
    }

    /// Writes a single record from string values.
    ///
    /// Each string is encoded using the configured encoding before writing.
    /// The `values` slice must contain one entry per field in the schema.
    /// To produce a padding-only column, pass an empty `""` value.
    ///
    /// # Errors
    ///
    /// Returns [`crate::FwfError::ValueCountMismatch`] if `values.len()` does not
    /// match the number of fields.
    ///
    /// Returns [`crate::FwfError::ValueTooLong`] if any encoded value exceeds its
    /// field width and the effective truncation policy is
    /// [`crate::TruncationPolicy::Error`].
    ///
    /// Returns [`crate::FwfError::Io`] on write failure.
    #[cfg(feature = "encoding_rs")]
    pub async fn write_string_record(&mut self, values: &[&str]) -> Result<()> {
        self.state.prepare_string_record(values)?;
        self.writer.write_all(self.state.record_buffer()).await?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use crate::fw_writer_field::FwWriterField;
    use crate::record_separator::RecordSeparator;
    use crate::{FwWriter, FwWriterSchema};

    #[tokio::test]
    async fn async_write_byte_records() {
        let schema = FwWriterSchema::builder()
            .field(FwWriterField::builder(5).build())
            .field(FwWriterField::builder(2).build())
            .separator(RecordSeparator::Lf)
            .build();
        let mut buf = Vec::new();
        let mut writer = FwWriter::options(schema).from_tokio_writer(&mut buf);

        writer.write_byte_record(&[b"Alice", b"30"]).await.unwrap();
        writer.write_byte_record(&[b"Bob", b"25"]).await.unwrap();
        assert_eq!(writer.record_number(), 2);
        writer.flush().await.unwrap();

        let writer = writer.into_inner();
        assert_eq!(writer, b"Alice30\nBob  25\n");
    }

    #[tokio::test]
    async fn async_write_with_padding() {
        let schema = FwWriterSchema::builder()
            .field(
                FwWriterField::builder(5)
                    .padding(b'0')
                    .alignment(crate::Alignment::Right)
                    .build(),
            )
            .build();
        let mut buf = Vec::new();
        let mut writer = FwWriter::options(schema).from_tokio_writer(&mut buf);

        writer.write_byte_record(&[b"42"]).await.unwrap();
        let writer = writer.into_inner();
        assert_eq!(writer, b"00042");
    }
}

#[cfg(test)]
#[cfg(feature = "encoding_rs")]
mod async_encoding_tests {
    use crate::fw_writer_field::FwWriterField;
    use crate::record_separator::RecordSeparator;
    use crate::{FwWriter, FwWriterSchema};

    #[tokio::test]
    async fn async_write_string_record() {
        let schema = FwWriterSchema::builder()
            .field(FwWriterField::builder(5).build())
            .field(FwWriterField::builder(2).build())
            .separator(RecordSeparator::Lf)
            .build();
        let mut buf = Vec::new();
        let mut writer = FwWriter::options(schema).from_tokio_writer(&mut buf);

        writer.write_string_record(&["Alice", "30"]).await.unwrap();
        writer.flush().await.unwrap();

        let writer = writer.into_inner();
        assert_eq!(writer, b"Alice30\n");
    }
}
