fastars 0.1.0

//! FASTQ file reading with automatic format detection.
//!
//! This module provides efficient FASTQ parsing using needletail,
//! with automatic detection of gzip compression based on file extension.
//!
//! ## Example
//!
//! ```no_run
//! use fastars::io::FastqReader;
//! use std::path::Path;
//!
//! let mut reader = FastqReader::new(Path::new("reads.fastq.gz")).unwrap();
//! let batch = reader.read_batch(1000).unwrap();
//! println!("Read {} records", batch.len());
//! ```

use std::fs::File;
use std::io::{BufReader, Read};
use std::path::Path;

use anyhow::{Context, Result};
use flate2::read::MultiGzDecoder;
use needletail::{parse_fastx_reader, FastxReader};

use super::pool::ReadPool;
use super::OwnedRecord;

/// Default batch size for reading records.
pub const DEFAULT_BATCH_SIZE: usize = 4096;

/// Buffer size for reading compressed files (1 MB).
/// Larger buffers improve performance with libdeflate/zlib-ng by reducing
/// syscalls and allowing more efficient decompression batches.
const BUFFER_SIZE: usize = 1024 * 1024;

/// Convert Phred64 quality scores to Phred33.
///
/// Phred64 uses ASCII 64-126 for quality scores, while Phred33 uses ASCII 33-126.
/// This function subtracts 31 from each quality byte to convert from Phred64 to Phred33.
#[inline]
fn convert_phred64_to_phred33(qual: &mut [u8]) {
    for q in qual.iter_mut() {
        *q = q.saturating_sub(31);
    }
}

/// Convert MGI sequencer read ID to Illumina-compatible format.
///
/// MGI format: @V350012345L1C001R001000001/1
/// - V350012345: Machine ID
/// - L1: Lane
/// - C001: Column
/// - R001: Row
/// - 000001: Tile/Read number
/// - /1: Read pair indicator
///
/// Converts to simplified Illumina-like format: @V350012345:L1:C001:R001:000001 1
fn convert_mgi_id(name: &[u8]) -> Vec<u8> {
    // Check if this looks like an MGI ID (starts with @ and contains pattern like LNCNNNRNNN)
    if name.is_empty() {
        return name.to_vec();
    }

    let name_str = std::str::from_utf8(name).unwrap_or("");

    // Check for MGI pattern: typically has L, C, R components
    if !name_str.contains('L') || !name_str.contains('C') || !name_str.contains('R') {
        return name.to_vec();
    }

    // Parse MGI format
    // Expected: V350012345L1C001R001000001/1 or similar patterns
    let mut result = Vec::with_capacity(name.len() + 10);

    // Find positions of L, C, R, and /
    if let Some(l_pos) = name_str.find('L') {
        if let Some(c_pos) = name_str[l_pos..].find('C').map(|p| p + l_pos) {
            if let Some(r_pos) = name_str[c_pos..].find('R').map(|p| p + c_pos) {
                // Machine ID (everything before L)
                result.extend_from_slice(&name[..l_pos]);
                result.push(b':');

                // Lane (L to C)
                result.extend_from_slice(&name[l_pos..c_pos]);
                result.push(b':');

                // Column (C to R)
                result.extend_from_slice(&name[c_pos..r_pos]);
                result.push(b':');

                // Row and rest (R to end or /)
                if let Some(slash_pos) = name_str[r_pos..].find('/').map(|p| p + r_pos) {
                    result.extend_from_slice(&name[r_pos..slash_pos]);
                    result.push(b' ');
                    result.extend_from_slice(&name[slash_pos + 1..]); // Read number after /
                } else {
                    result.extend_from_slice(&name[r_pos..]);
                }

                return result;
            }
        }
    }

    // If parsing failed, return original
    name.to_vec()
}

/// Buffer size for the inner file reader (256 KB).
/// Smaller than the decompression buffer since raw I/O benefits less from
/// very large buffers.
const FILE_BUFFER_SIZE: usize = 256 * 1024;

/// Determines if a file path indicates gzip compression.
fn is_gzipped(path: &Path) -> bool {
    let path_str = path.to_string_lossy().to_lowercase();
    path_str.ends_with(".gz") || path_str.ends_with(".gzip")
}

/// Detect if stdin contains gzipped data by reading magic bytes.
///
/// Returns (is_gzipped, first_bytes) where first_bytes is the buffered data
/// that needs to be prepended to the stream.
fn detect_gzip_stdin<R: Read>(
    reader: &mut R,
) -> Result<(bool, Vec<u8>)> {
    let mut magic = [0u8; 2];
    match reader.read(&mut magic) {
        Ok(0) => Ok((false, Vec::new())), // Empty stream
        Ok(1) => Ok((false, vec![magic[0]])), // Single byte, not gzip
        Ok(2) => {
            let is_gzip = magic[0] == 0x1f && magic[1] == 0x8b;
            Ok((is_gzip, magic.to_vec()))
        }
        Ok(_) => unreachable!(),
        Err(e) => Err(anyhow::anyhow!("Failed to read from stdin: {}", e)),
    }
}

/// Create a FASTQ reader from stdin with optional format detection.
///
/// # Arguments
///
/// * `auto_detect` - If true, detect gzip compression from magic bytes.
///                   If false, assume plain text input.
/// * `force_gzip` - If true, force gzip decompression regardless of magic bytes.
///
/// # Errors
///
/// Returns an error if stdin cannot be read or parsed.
pub fn create_stdin_reader(auto_detect: bool, force_gzip: bool) -> Result<FastqReader> {
    use std::io::{stdin, Cursor};

    let stdin_handle = stdin();
    let mut stdin_reader = BufReader::with_capacity(FILE_BUFFER_SIZE, stdin_handle);

    let reader: Box<dyn Read + Send> = if force_gzip {
        // Force gzip decompression
        Box::new(BufReader::with_capacity(
            BUFFER_SIZE,
            MultiGzDecoder::new(stdin_reader),
        ))
    } else if auto_detect {
        // Auto-detect format from magic bytes
        let (is_gzip, magic_bytes) = detect_gzip_stdin(&mut stdin_reader)?;

        if is_gzip {
            // Prepend magic bytes and decompress
            let chained = Cursor::new(magic_bytes).chain(stdin_reader);
            Box::new(BufReader::with_capacity(
                BUFFER_SIZE,
                MultiGzDecoder::new(chained),
            ))
        } else {
            // Plain text: prepend magic bytes
            let chained = Cursor::new(magic_bytes).chain(stdin_reader);
            Box::new(BufReader::with_capacity(BUFFER_SIZE, chained))
        }
    } else {
        // Plain text, no detection
        Box::new(BufReader::with_capacity(BUFFER_SIZE, stdin_reader))
    };

    let fastx_reader = parse_fastx_reader(reader)
        .context("Failed to parse FASTQ from stdin")?;

    Ok(FastqReader {
        reader: fastx_reader,
        finished: false,
        convert_phred64: false,
        fix_mgi_id: false,
    })
}

/// FASTQ file reader with automatic format detection.
///
/// Supports both plain text and gzip-compressed FASTQ files.
/// Compression is auto-detected based on file extension (.gz, .gzip).
pub struct FastqReader {
    reader: Box<dyn FastxReader>,
    finished: bool,
    /// Convert Phred64 to Phred33 on read
    convert_phred64: bool,
    /// Fix MGI read IDs to Illumina-compatible format
    fix_mgi_id: bool,
}

impl FastqReader {
    /// Create a new reader for the given file path.
    ///
    /// Automatically detects gzip compression from file extension:
    /// - `.fq`, `.fastq` - plain text
    /// - `.fq.gz`, `.fastq.gz` - gzip compressed
    ///
    /// # Errors
    ///
    /// Returns an error if the file cannot be opened or parsed.
    pub fn new(path: &Path) -> Result<Self> {
        let file = File::open(path)
            .with_context(|| format!("Failed to open file: {}", path.display()))?;

        let reader: Box<dyn Read + Send> = if is_gzipped(path) {
            // For gzip: inner buffer for file I/O, outer buffer for decompressed data
            Box::new(BufReader::with_capacity(
                BUFFER_SIZE,
                MultiGzDecoder::new(BufReader::with_capacity(FILE_BUFFER_SIZE, file)),
            ))
        } else {
            // For plain text: single buffer is sufficient
            Box::new(BufReader::with_capacity(BUFFER_SIZE, file))
        };

        let fastx_reader = parse_fastx_reader(reader)
            .with_context(|| format!("Failed to parse FASTQ from: {}", path.display()))?;

        Ok(Self {
            reader: fastx_reader,
            finished: false,
            convert_phred64: false,
            fix_mgi_id: false,
        })
    }

    /// Set Phred64 conversion mode.
    ///
    /// When enabled, quality scores are converted from Phred64 (ASCII 64-126)
    /// to Phred33 (ASCII 33-126) by subtracting 31 from each quality byte.
    pub fn with_phred64_conversion(mut self, convert: bool) -> Self {
        self.convert_phred64 = convert;
        self
    }

    /// Set MGI ID conversion mode.
    ///
    /// When enabled, read IDs are converted from MGI format to Illumina-compatible format.
    pub fn with_mgi_id_conversion(mut self, convert: bool) -> Self {
        self.fix_mgi_id = convert;
        self
    }

    /// Read a batch of records from the file.
    ///
    /// Returns up to `batch_size` records. Returns an empty vector when
    /// the end of file is reached.
    ///
    /// # Arguments
    ///
    /// * `batch_size` - Maximum number of records to read (default: 4096)
    ///
    /// # Errors
    ///
    /// Returns an error if there's a problem reading or parsing records.
    pub fn read_batch(&mut self, batch_size: usize) -> Result<Vec<OwnedRecord>> {
        if self.finished {
            return Ok(Vec::new());
        }

        let mut records = Vec::with_capacity(batch_size);

        while records.len() < batch_size {
            match self.reader.next() {
                Some(Ok(record)) => {
                    let mut qual = record.qual().map(|q| q.to_vec()).unwrap_or_default();
                    if self.convert_phred64 && !qual.is_empty() {
                        convert_phred64_to_phred33(&mut qual);
                    }
                    let name = if self.fix_mgi_id {
                        convert_mgi_id(record.id())
                    } else {
                        record.id().to_vec()
                    };
                    let owned = OwnedRecord::new(
                        name,
                        record.seq().to_vec(),
                        qual,
                    );
                    records.push(owned);
                }
                Some(Err(e)) => {
                    return Err(anyhow::anyhow!("Error parsing FASTQ record: {}", e));
                }
                None => {
                    self.finished = true;
                    break;
                }
            }
        }

        Ok(records)
    }

    /// Read a batch of records using a pool for memory reuse.
    ///
    /// This method uses the provided `ReadPool` to acquire records instead
    /// of allocating new ones, significantly reducing heap allocations when
    /// processing large files.
    ///
    /// # Arguments
    ///
    /// * `batch_size` - Maximum number of records to read
    /// * `pool` - The pool to acquire records from
    ///
    /// # Errors
    ///
    /// Returns an error if there's a problem reading or parsing records.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use fastars::io::{FastqReader, ReadPool};
    ///
    /// let mut reader = FastqReader::new(Path::new("reads.fastq")).unwrap();
    /// let mut pool = ReadPool::new(256);
    ///
    /// loop {
    ///     let batch = reader.read_batch_pooled(1000, &mut pool).unwrap();
    ///     if batch.is_empty() {
    ///         break;
    ///     }
    ///     // Process batch...
    ///     pool.release_batch(batch); // Return records to pool
    /// }
    /// ```
    pub fn read_batch_pooled(
        &mut self,
        batch_size: usize,
        pool: &mut ReadPool,
    ) -> Result<Vec<OwnedRecord>> {
        if self.finished {
            return Ok(Vec::new());
        }

        let mut records = Vec::with_capacity(batch_size);

        while records.len() < batch_size {
            match self.reader.next() {
                Some(Ok(record)) => {
                    let mut owned = pool.acquire();
                    let qual = record.qual().unwrap_or(&[]);
                    let name = if self.fix_mgi_id {
                        convert_mgi_id(record.id())
                    } else {
                        record.id().to_vec()
                    };
                    if self.convert_phred64 && !qual.is_empty() {
                        // Need to convert before setting
                        let mut qual_vec = qual.to_vec();
                        convert_phred64_to_phred33(&mut qual_vec);
                        owned.set_from(
                            &name,
                            &record.seq(),
                            &qual_vec,
                        );
                    } else {
                        owned.set_from(
                            &name,
                            &record.seq(),
                            qual,
                        );
                    }
                    records.push(owned);
                }
                Some(Err(e)) => {
                    // Return already-read records to pool before returning error
                    pool.release_batch(records);
                    return Err(anyhow::anyhow!("Error parsing FASTQ record: {}", e));
                }
                None => {
                    self.finished = true;
                    break;
                }
            }
        }

        Ok(records)
    }

    /// Read into an existing record, reusing its memory.
    ///
    /// This method fills the provided record with data from the next read,
    /// avoiding allocation by reusing the record's existing capacity.
    ///
    /// Returns `true` if a record was read, `false` if EOF reached.
    ///
    /// # Arguments
    ///
    /// * `record` - The record to fill with data
    ///
    /// # Errors
    ///
    /// Returns an error if there's a problem reading or parsing the record.
    pub fn read_into(&mut self, record: &mut OwnedRecord) -> Result<bool> {
        if self.finished {
            return Ok(false);
        }

        match self.reader.next() {
            Some(Ok(rec)) => {
                let qual = rec.qual().unwrap_or(&[]);
                let name = if self.fix_mgi_id {
                    convert_mgi_id(rec.id())
                } else {
                    rec.id().to_vec()
                };
                if self.convert_phred64 && !qual.is_empty() {
                    // Need to convert before setting
                    let mut qual_vec = qual.to_vec();
                    convert_phred64_to_phred33(&mut qual_vec);
                    record.set_from(
                        &name,
                        &rec.seq(),
                        &qual_vec,
                    );
                } else {
                    record.set_from(
                        &name,
                        &rec.seq(),
                        qual,
                    );
                }
                Ok(true)
            }
            Some(Err(e)) => Err(anyhow::anyhow!("Error parsing FASTQ record: {}", e)),
            None => {
                self.finished = true;
                Ok(false)
            }
        }
    }

    /// Read a batch of interleaved paired-end records.
    ///
    /// Reads alternating R1/R2 records from an interleaved file.
    /// Returns two vectors: (R1 records, R2 records).
    ///
    /// # Arguments
    ///
    /// * `batch_size` - Maximum number of record pairs to read
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - There's a problem reading or parsing records
    /// - The file has an odd number of records (incomplete pair)
    pub fn read_batch_interleaved(&mut self, batch_size: usize) -> Result<(Vec<OwnedRecord>, Vec<OwnedRecord>)> {
        if self.finished {
            return Ok((Vec::new(), Vec::new()));
        }

        let mut r1_records = Vec::with_capacity(batch_size);
        let mut r2_records = Vec::with_capacity(batch_size);

        while r1_records.len() < batch_size {
            // Read R1
            match self.reader.next() {
                Some(Ok(record)) => {
                    let mut qual = record.qual().map(|q| q.to_vec()).unwrap_or_default();
                    if self.convert_phred64 && !qual.is_empty() {
                        convert_phred64_to_phred33(&mut qual);
                    }
                    let name = if self.fix_mgi_id {
                        convert_mgi_id(record.id())
                    } else {
                        record.id().to_vec()
                    };
                    let owned = OwnedRecord::new(
                        name,
                        record.seq().to_vec(),
                        qual,
                    );
                    r1_records.push(owned);
                }
                Some(Err(e)) => {
                    return Err(anyhow::anyhow!("Error parsing R1 record: {}", e));
                }
                None => {
                    self.finished = true;
                    if !r2_records.is_empty() && r1_records.len() != r2_records.len() {
                        return Err(anyhow::anyhow!("Interleaved file has incomplete pair at end"));
                    }
                    break;
                }
            }

            // Read R2
            match self.reader.next() {
                Some(Ok(record)) => {
                    let mut qual = record.qual().map(|q| q.to_vec()).unwrap_or_default();
                    if self.convert_phred64 && !qual.is_empty() {
                        convert_phred64_to_phred33(&mut qual);
                    }
                    let name = if self.fix_mgi_id {
                        convert_mgi_id(record.id())
                    } else {
                        record.id().to_vec()
                    };
                    let owned = OwnedRecord::new(
                        name,
                        record.seq().to_vec(),
                        qual,
                    );
                    r2_records.push(owned);
                }
                Some(Err(e)) => {
                    return Err(anyhow::anyhow!("Error parsing R2 record: {}", e));
                }
                None => {
                    self.finished = true;
                    return Err(anyhow::anyhow!("Interleaved file has incomplete pair at end"));
                }
            }
        }

        Ok((r1_records, r2_records))
    }

    /// Read a batch of interleaved paired-end records using pools for memory reuse.
    ///
    /// Reads alternating R1/R2 records from an interleaved file.
    /// Returns pairs of (R1, R2) records matching PairedFastqReader format.
    ///
    /// # Arguments
    ///
    /// * `batch_size` - Maximum number of record pairs to read
    /// * `pool1` - The pool to acquire R1 records from
    /// * `pool2` - The pool to acquire R2 records from
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - There's a problem reading or parsing records
    /// - The file has an odd number of records (incomplete pair)
    pub fn read_batch_interleaved_pooled(
        &mut self,
        batch_size: usize,
        pool1: &mut ReadPool,
        pool2: &mut ReadPool,
    ) -> Result<Vec<(OwnedRecord, OwnedRecord)>> {
        if self.finished {
            return Ok(Vec::new());
        }

        let mut pairs = Vec::with_capacity(batch_size);

        while pairs.len() < batch_size {
            // Read R1
            let r1 = match self.reader.next() {
                Some(Ok(record)) => {
                    let mut owned = pool1.acquire();
                    let qual = record.qual().unwrap_or(&[]);
                    let name = if self.fix_mgi_id {
                        convert_mgi_id(record.id())
                    } else {
                        record.id().to_vec()
                    };
                    if self.convert_phred64 && !qual.is_empty() {
                        let mut qual_vec = qual.to_vec();
                        convert_phred64_to_phred33(&mut qual_vec);
                        owned.set_from(
                            &name,
                            &record.seq(),
                            &qual_vec,
                        );
                    } else {
                        owned.set_from(
                            &name,
                            &record.seq(),
                            qual,
                        );
                    }
                    owned
                }
                Some(Err(e)) => {
                    // Return already-read records to pools before returning error
                    for (r1, r2) in pairs {
                        pool1.release(r1);
                        pool2.release(r2);
                    }
                    return Err(anyhow::anyhow!("Error parsing R1 record: {}", e));
                }
                None => {
                    self.finished = true;
                    break;
                }
            };

            // Read R2
            let r2 = match self.reader.next() {
                Some(Ok(record)) => {
                    let mut owned = pool2.acquire();
                    let qual = record.qual().unwrap_or(&[]);
                    let name = if self.fix_mgi_id {
                        convert_mgi_id(record.id())
                    } else {
                        record.id().to_vec()
                    };
                    if self.convert_phred64 && !qual.is_empty() {
                        let mut qual_vec = qual.to_vec();
                        convert_phred64_to_phred33(&mut qual_vec);
                        owned.set_from(
                            &name,
                            &record.seq(),
                            &qual_vec,
                        );
                    } else {
                        owned.set_from(
                            &name,
                            &record.seq(),
                            qual,
                        );
                    }
                    owned
                }
                Some(Err(e)) => {
                    // Return R1 and all previous pairs to pools before returning error
                    pool1.release(r1);
                    for (r1, r2) in pairs {
                        pool1.release(r1);
                        pool2.release(r2);
                    }
                    return Err(anyhow::anyhow!("Error parsing R2 record: {}", e));
                }
                None => {
                    self.finished = true;
                    // Odd number of records - return R1 and all pairs to pools
                    pool1.release(r1);
                    for (r1, r2) in pairs {
                        pool1.release(r1);
                        pool2.release(r2);
                    }
                    return Err(anyhow::anyhow!("Interleaved file has incomplete pair at end"));
                }
            };

            pairs.push((r1, r2));
        }

        Ok(pairs)
    }

    /// Read the next single record from the file.
    ///
    /// Returns `None` when the end of file is reached.
    ///
    /// # Errors
    ///
    /// Returns an error if there's a problem reading or parsing the record.
    pub fn read_record(&mut self) -> Result<Option<OwnedRecord>> {
        if self.finished {
            return Ok(None);
        }

        match self.reader.next() {
            Some(Ok(record)) => {
                let mut qual = record.qual().map(|q| q.to_vec()).unwrap_or_default();
                if self.convert_phred64 && !qual.is_empty() {
                    convert_phred64_to_phred33(&mut qual);
                }
                let name = if self.fix_mgi_id {
                    convert_mgi_id(record.id())
                } else {
                    record.id().to_vec()
                };
                let owned = OwnedRecord::new(
                    name,
                    record.seq().to_vec(),
                    qual,
                );
                Ok(Some(owned))
            }
            Some(Err(e)) => Err(anyhow::anyhow!("Error parsing FASTQ record: {}", e)),
            None => {
                self.finished = true;
                Ok(None)
            }
        }
    }

    /// Check if the reader has finished reading all records.
    pub fn is_finished(&self) -> bool {
        self.finished
    }
}

/// Iterator implementation for FastqReader.
impl Iterator for FastqReader {
    type Item = Result<OwnedRecord>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.finished {
            return None;
        }

        match self.reader.next() {
            Some(Ok(record)) => {
                let mut qual = record.qual().map(|q| q.to_vec()).unwrap_or_default();
                if self.convert_phred64 && !qual.is_empty() {
                    convert_phred64_to_phred33(&mut qual);
                }
                let name = if self.fix_mgi_id {
                    convert_mgi_id(record.id())
                } else {
                    record.id().to_vec()
                };
                let owned = OwnedRecord::new(
                    name,
                    record.seq().to_vec(),
                    qual,
                );
                Some(Ok(owned))
            }
            Some(Err(e)) => Some(Err(anyhow::anyhow!("Error parsing FASTQ record: {}", e))),
            None => {
                self.finished = true;
                None
            }
        }
    }
}

/// Synchronized reader for paired-end FASTQ files.
///
/// Ensures that R1 and R2 records are read in lockstep, maintaining
/// proper pairing between forward and reverse reads.
pub struct PairedFastqReader {
    reader1: FastqReader,
    reader2: FastqReader,
}

impl PairedFastqReader {
    /// Create a new paired-end reader for R1 and R2 files.
    ///
    /// # Arguments
    ///
    /// * `path1` - Path to the R1 (forward) FASTQ file
    /// * `path2` - Path to the R2 (reverse) FASTQ file
    ///
    /// # Errors
    ///
    /// Returns an error if either file cannot be opened.
    pub fn new(path1: &Path, path2: &Path) -> Result<Self> {
        let reader1 = FastqReader::new(path1)
            .with_context(|| format!("Failed to open R1 file: {}", path1.display()))?;
        let reader2 = FastqReader::new(path2)
            .with_context(|| format!("Failed to open R2 file: {}", path2.display()))?;

        Ok(Self { reader1, reader2 })
    }

    /// Set Phred64 conversion mode for both readers.
    ///
    /// When enabled, quality scores are converted from Phred64 (ASCII 64-126)
    /// to Phred33 (ASCII 33-126) by subtracting 31 from each quality byte.
    pub fn with_phred64_conversion(mut self, convert: bool) -> Self {
        self.reader1.convert_phred64 = convert;
        self.reader2.convert_phred64 = convert;
        self
    }

    /// Set MGI ID conversion mode for both readers.
    ///
    /// When enabled, read IDs are converted from MGI format to Illumina-compatible format.
    pub fn with_mgi_id_conversion(mut self, convert: bool) -> Self {
        self.reader1.fix_mgi_id = convert;
        self.reader2.fix_mgi_id = convert;
        self
    }

    /// Read a batch of paired records from both files.
    ///
    /// Returns pairs of (R1, R2) records. If the files have different
    /// numbers of records, returns an error when one file ends before the other.
    ///
    /// # Arguments
    ///
    /// * `batch_size` - Maximum number of pairs to read (default: 4096)
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - There's a problem reading or parsing records
    /// - Files have different numbers of records (detected during reading)
    pub fn read_batch(&mut self, batch_size: usize) -> Result<Vec<(OwnedRecord, OwnedRecord)>> {
        let mut pairs = Vec::with_capacity(batch_size);

        loop {
            if pairs.len() >= batch_size {
                break;
            }

            let rec1 = self.reader1.read_record()?;
            let rec2 = self.reader2.read_record()?;

            match (rec1, rec2) {
                (Some(r1), Some(r2)) => {
                    pairs.push((r1, r2));
                }
                (None, None) => {
                    // Both files ended at the same time - good
                    break;
                }
                (Some(_), None) => {
                    return Err(anyhow::anyhow!(
                        "R1 file has more records than R2 file"
                    ));
                }
                (None, Some(_)) => {
                    return Err(anyhow::anyhow!(
                        "R2 file has more records than R1 file"
                    ));
                }
            }
        }

        Ok(pairs)
    }

    /// Read a batch of paired records using pools for memory reuse.
    ///
    /// This method uses the provided `ReadPool`s to acquire records instead
    /// of allocating new ones, significantly reducing heap allocations when
    /// processing large paired-end files.
    ///
    /// # Arguments
    ///
    /// * `batch_size` - Maximum number of pairs to read
    /// * `pool1` - The pool to acquire R1 records from
    /// * `pool2` - The pool to acquire R2 records from
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - There's a problem reading or parsing records
    /// - Files have different numbers of records
    pub fn read_batch_pooled(
        &mut self,
        batch_size: usize,
        pool1: &mut ReadPool,
        pool2: &mut ReadPool,
    ) -> Result<Vec<(OwnedRecord, OwnedRecord)>> {
        let mut pairs = Vec::with_capacity(batch_size);

        loop {
            if pairs.len() >= batch_size {
                break;
            }

            // Acquire records from pools
            let mut r1 = pool1.acquire();
            let mut r2 = pool2.acquire();

            let has_r1 = self.reader1.read_into(&mut r1)?;
            let has_r2 = self.reader2.read_into(&mut r2)?;

            match (has_r1, has_r2) {
                (true, true) => {
                    pairs.push((r1, r2));
                }
                (false, false) => {
                    // Both files ended - return unused records to pool
                    pool1.release(r1);
                    pool2.release(r2);
                    break;
                }
                (true, false) => {
                    // Release records and return error
                    pool1.release(r1);
                    pool2.release(r2);
                    return Err(anyhow::anyhow!(
                        "R1 file has more records than R2 file"
                    ));
                }
                (false, true) => {
                    // Release records and return error
                    pool1.release(r1);
                    pool2.release(r2);
                    return Err(anyhow::anyhow!(
                        "R2 file has more records than R1 file"
                    ));
                }
            }
        }

        Ok(pairs)
    }

    /// Read the next pair of records from both files.
    ///
    /// Returns `None` when both files end. Returns an error if files
    /// have different numbers of records.
    pub fn read_pair(&mut self) -> Result<Option<(OwnedRecord, OwnedRecord)>> {
        let rec1 = self.reader1.read_record()?;
        let rec2 = self.reader2.read_record()?;

        match (rec1, rec2) {
            (Some(r1), Some(r2)) => Ok(Some((r1, r2))),
            (None, None) => Ok(None),
            (Some(_), None) => Err(anyhow::anyhow!(
                "R1 file has more records than R2 file"
            )),
            (None, Some(_)) => Err(anyhow::anyhow!(
                "R2 file has more records than R1 file"
            )),
        }
    }

    /// Check if both readers have finished.
    pub fn is_finished(&self) -> bool {
        self.reader1.is_finished() && self.reader2.is_finished()
    }
}

/// Iterator implementation for PairedFastqReader.
impl Iterator for PairedFastqReader {
    type Item = Result<(OwnedRecord, OwnedRecord)>;

    fn next(&mut self) -> Option<Self::Item> {
        match self.read_pair() {
            Ok(Some(pair)) => Some(Ok(pair)),
            Ok(None) => None,
            Err(e) => Some(Err(e)),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Write;
    use tempfile::NamedTempFile;

    fn create_temp_fastq(contents: &[u8]) -> NamedTempFile {
        let mut file = NamedTempFile::with_suffix(".fastq").unwrap();
        file.write_all(contents).unwrap();
        file.flush().unwrap();
        file
    }

    fn create_temp_fastq_gz(contents: &[u8]) -> NamedTempFile {
        use flate2::write::GzEncoder;
        use flate2::Compression;

        let mut file = NamedTempFile::with_suffix(".fastq.gz").unwrap();
        {
            let mut encoder = GzEncoder::new(&mut file, Compression::default());
            encoder.write_all(contents).unwrap();
            encoder.finish().unwrap();
        }
        file.flush().unwrap();
        file
    }

    const SAMPLE_FASTQ: &[u8] = b"@read1
ACGTACGT
+
IIIIIIII
@read2
TGCATGCA
+
HHHHHHHH
";

    #[test]
    fn test_reader_plain_text() {
        let file = create_temp_fastq(SAMPLE_FASTQ);
        let mut reader = FastqReader::new(file.path()).unwrap();

        let records = reader.read_batch(10).unwrap();
        assert_eq!(records.len(), 2);
        assert_eq!(records[0].name, b"read1");
        assert_eq!(records[0].seq, b"ACGTACGT");
        assert_eq!(records[0].qual, b"IIIIIIII");
        assert_eq!(records[1].name, b"read2");
    }

    #[test]
    fn test_reader_gzipped() {
        let file = create_temp_fastq_gz(SAMPLE_FASTQ);
        let mut reader = FastqReader::new(file.path()).unwrap();

        let records = reader.read_batch(10).unwrap();
        assert_eq!(records.len(), 2);
        assert_eq!(records[0].name, b"read1");
        assert_eq!(records[0].seq, b"ACGTACGT");
    }

    #[test]
    fn test_reader_batch_size() {
        let file = create_temp_fastq(SAMPLE_FASTQ);
        let mut reader = FastqReader::new(file.path()).unwrap();

        // Read just one record
        let batch1 = reader.read_batch(1).unwrap();
        assert_eq!(batch1.len(), 1);
        assert_eq!(batch1[0].name, b"read1");

        // Read the remaining record
        let batch2 = reader.read_batch(1).unwrap();
        assert_eq!(batch2.len(), 1);
        assert_eq!(batch2[0].name, b"read2");

        // No more records
        let batch3 = reader.read_batch(1).unwrap();
        assert!(batch3.is_empty());
    }

    #[test]
    fn test_reader_iterator() {
        let file = create_temp_fastq(SAMPLE_FASTQ);
        let reader = FastqReader::new(file.path()).unwrap();

        let records: Vec<_> = reader.map(|r| r.unwrap()).collect();
        assert_eq!(records.len(), 2);
        assert_eq!(records[0].name, b"read1");
        assert_eq!(records[1].name, b"read2");
    }

    #[test]
    fn test_reader_empty_file() {
        let file = create_temp_fastq(b"");
        // needletail requires at least 2 bytes to detect file format
        // Empty files fail at parse time, which is expected behavior
        let result = FastqReader::new(file.path());
        assert!(result.is_err());
    }

    #[test]
    fn test_paired_reader() {
        let r1_content = b"@read1/1
AAAA
+
IIII
@read2/1
CCCC
+
IIII
";
        let r2_content = b"@read1/2
TTTT
+
IIII
@read2/2
GGGG
+
IIII
";
        let file1 = create_temp_fastq(r1_content);
        let file2 = create_temp_fastq(r2_content);

        let mut reader = PairedFastqReader::new(file1.path(), file2.path()).unwrap();
        let pairs = reader.read_batch(10).unwrap();

        assert_eq!(pairs.len(), 2);
        assert_eq!(pairs[0].0.seq, b"AAAA");
        assert_eq!(pairs[0].1.seq, b"TTTT");
        assert_eq!(pairs[1].0.seq, b"CCCC");
        assert_eq!(pairs[1].1.seq, b"GGGG");
    }

    #[test]
    fn test_paired_reader_mismatch() {
        let r1_content = b"@read1/1
AAAA
+
IIII
@read2/1
CCCC
+
IIII
";
        let r2_content = b"@read1/2
TTTT
+
IIII
";
        let file1 = create_temp_fastq(r1_content);
        let file2 = create_temp_fastq(r2_content);

        let mut reader = PairedFastqReader::new(file1.path(), file2.path()).unwrap();
        let result = reader.read_batch(10);

        assert!(result.is_err());
        assert!(result
            .unwrap_err()
            .to_string()
            .contains("more records"));
    }

    #[test]
    fn test_is_gzipped_detection() {
        assert!(is_gzipped(Path::new("file.fastq.gz")));
        assert!(is_gzipped(Path::new("file.fq.gz")));
        assert!(is_gzipped(Path::new("file.FASTQ.GZ")));
        assert!(is_gzipped(Path::new("file.gzip")));
        assert!(!is_gzipped(Path::new("file.fastq")));
        assert!(!is_gzipped(Path::new("file.fq")));
    }

    #[test]
    fn test_read_record() {
        let file = create_temp_fastq(SAMPLE_FASTQ);
        let mut reader = FastqReader::new(file.path()).unwrap();

        let rec1 = reader.read_record().unwrap();
        assert!(rec1.is_some());
        assert_eq!(rec1.unwrap().name, b"read1");

        let rec2 = reader.read_record().unwrap();
        assert!(rec2.is_some());
        assert_eq!(rec2.unwrap().name, b"read2");

        let rec3 = reader.read_record().unwrap();
        assert!(rec3.is_none());
    }

    #[test]
    fn test_read_batch_pooled() {
        let file = create_temp_fastq(SAMPLE_FASTQ);
        let mut reader = FastqReader::new(file.path()).unwrap();
        let mut pool = ReadPool::new(256);

        let records = reader.read_batch_pooled(10, &mut pool).unwrap();
        assert_eq!(records.len(), 2);
        assert_eq!(records[0].name, b"read1");
        assert_eq!(records[0].seq, b"ACGTACGT");
        assert_eq!(records[0].qual, b"IIIIIIII");
        assert_eq!(records[1].name, b"read2");

        // Return records to pool
        pool.release_batch(records);
        assert_eq!(pool.len(), 2);

        // Read again (should be empty now)
        let records2 = reader.read_batch_pooled(10, &mut pool).unwrap();
        assert!(records2.is_empty());
    }

    #[test]
    fn test_read_into() {
        let file = create_temp_fastq(SAMPLE_FASTQ);
        let mut reader = FastqReader::new(file.path()).unwrap();
        let mut record = OwnedRecord::with_capacity(256);

        // Read first record
        assert!(reader.read_into(&mut record).unwrap());
        assert_eq!(record.name, b"read1");
        assert_eq!(record.seq, b"ACGTACGT");

        // Read second record (reuses memory)
        assert!(reader.read_into(&mut record).unwrap());
        assert_eq!(record.name, b"read2");
        assert_eq!(record.seq, b"TGCATGCA");

        // No more records
        assert!(!reader.read_into(&mut record).unwrap());
    }

    #[test]
    fn test_pool_memory_reuse() {
        let file = create_temp_fastq(SAMPLE_FASTQ);
        let mut reader = FastqReader::new(file.path()).unwrap();
        let mut pool = ReadPool::new(256);

        // Pre-fill the pool
        pool.prefill(5);
        assert_eq!(pool.len(), 5);

        // Read records - should use pooled records
        let records = reader.read_batch_pooled(2, &mut pool).unwrap();
        assert_eq!(records.len(), 2);
        assert_eq!(pool.len(), 3); // 5 - 2 = 3 remaining

        // Return to pool
        pool.release_batch(records);
        assert_eq!(pool.len(), 5); // Back to 5
    }

    #[test]
    fn test_paired_read_batch_pooled() {
        let r1_content = b"@read1/1
AAAA
+
IIII
@read2/1
CCCC
+
IIII
";
        let r2_content = b"@read1/2
TTTT
+
IIII
@read2/2
GGGG
+
IIII
";
        let file1 = create_temp_fastq(r1_content);
        let file2 = create_temp_fastq(r2_content);

        let mut reader = PairedFastqReader::new(file1.path(), file2.path()).unwrap();
        let mut pool1 = ReadPool::new(256);
        let mut pool2 = ReadPool::new(256);

        let pairs = reader.read_batch_pooled(10, &mut pool1, &mut pool2).unwrap();

        assert_eq!(pairs.len(), 2);
        assert_eq!(pairs[0].0.seq, b"AAAA");
        assert_eq!(pairs[0].1.seq, b"TTTT");
        assert_eq!(pairs[1].0.seq, b"CCCC");
        assert_eq!(pairs[1].1.seq, b"GGGG");
    }
}