// Dweve HEDL - Hierarchical Entity Data Language
//
// Copyright (c) 2025 Dweve IP B.V. and individual contributors.
//
// SPDX-License-Identifier: Apache-2.0
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License in the LICENSE file at the
// root of this repository or at: http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Async line reader for streaming parser.
//!
//! Provides buffered async line-by-line reading with line number tracking, peek support,
//! and the ability to push back lines for re-parsing.
//!
//! This module mirrors the synchronous [`LineReader`](crate::LineReader) but uses
//! tokio's async I/O primitives for non-blocking operation.
//!
//! # Examples
//!
//! ## Basic Async Line Reading
//!
//! ```rust,no_run
//! # #[cfg(feature = "async")]
//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
//! use hedl_stream::AsyncLineReader;
//! use tokio::io::AsyncReadExt;
//! use std::io::Cursor;
//!
//! let input = "line1\nline2\nline3";
//! let mut reader = AsyncLineReader::new(Cursor::new(input));
//!
//! assert_eq!(reader.next_line().await?, Some((1, "line1".to_string())));
//! assert_eq!(reader.next_line().await?, Some((2, "line2".to_string())));
//! assert_eq!(reader.next_line().await?, Some((3, "line3".to_string())));
//! assert_eq!(reader.next_line().await?, None);
//! # Ok(())
//! # }
//! ```
//!
//! ## Peeking and Push Back
//!
//! ```rust,no_run
//! # #[cfg(feature = "async")]
//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
//! use hedl_stream::AsyncLineReader;
//! use std::io::Cursor;
//!
//! let input = "line1\nline2";
//! let mut reader = AsyncLineReader::new(Cursor::new(input));
//!
//! // Peek without consuming
//! assert_eq!(reader.peek_line().await?, Some(&(1, "line1".to_string())));
//! assert_eq!(reader.peek_line().await?, Some(&(1, "line1".to_string())));
//!
//! // Now consume it
//! let line = reader.next_line().await?.unwrap();
//! assert_eq!(line, (1, "line1".to_string()));
//!
//! // Push it back
//! reader.push_back(line.0, line.1);
//!
//! // Read it again
//! assert_eq!(reader.next_line().await?, Some((1, "line1".to_string())));
//! # Ok(())
//! # }
//! ```

use crate::error::{StreamError, StreamResult};
use tokio::io::{AsyncBufReadExt, AsyncRead, BufReader};

/// Buffered async line reader with line number tracking.
///
/// Reads input line-by-line asynchronously, automatically handling different line endings
/// (LF, CRLF) and tracking the current line number for error reporting.
///
/// # Performance Characteristics
///
/// - **Buffering**: Configurable buffer size (default 64KB) for efficient I/O
/// - **Zero-Copy**: String allocations only for consumed lines
/// - **Async**: Non-blocking I/O suitable for high-concurrency scenarios
///
/// # When to Use Async vs Sync
///
/// **Use Async When:**
/// - Processing network streams or pipes
/// - High-concurrency scenarios (many parallel streams)
/// - Integration with async web servers or frameworks
/// - Need to process I/O without blocking threads
///
/// **Use Sync When:**
/// - Processing local files
/// - Single-threaded batch processing
/// - Simpler code without async complexity
/// - CPU-bound workloads with minimal I/O wait
///
/// # Examples
///
/// ## Reading from Async Source
///
/// ```rust,no_run
/// # #[cfg(feature = "async")]
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// use hedl_stream::AsyncLineReader;
/// use tokio::fs::File;
///
/// let file = File::open("data.hedl").await?;
/// let mut reader = AsyncLineReader::new(file);
///
/// while let Some((line_num, line)) = reader.next_line().await? {
///     println!("{}: {}", line_num, line);
/// }
/// # Ok(())
/// # }
/// ```
///
/// ## With Custom Buffer Size
///
/// ```rust
/// # #[cfg(feature = "async")]
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// use hedl_stream::AsyncLineReader;
/// use std::io::Cursor;
///
/// let input = "line1\nline2";
/// let reader = AsyncLineReader::with_capacity(Cursor::new(input), 256 * 1024);
/// # Ok(())
/// # }
/// ```
pub struct AsyncLineReader<R: AsyncRead + Unpin> {
    reader: BufReader<R>,
    line_number: usize,
    buffer: String,
    peeked: Option<(usize, String)>,
}

impl<R: AsyncRead + Unpin> AsyncLineReader<R> {
    /// Create a new async line reader with default buffer size (64KB).
    ///
    /// # Examples
    ///
    /// ```rust
    /// # #[cfg(feature = "async")]
    /// use hedl_stream::AsyncLineReader;
    /// use std::io::Cursor;
    ///
    /// let input = "line1\nline2";
    /// let reader = AsyncLineReader::new(Cursor::new(input));
    /// ```
    pub fn new(reader: R) -> Self {
        Self {
            reader: BufReader::new(reader),
            line_number: 0,
            buffer: String::new(),
            peeked: None,
        }
    }

    /// Create an async line reader with a specific buffer capacity.
    ///
    /// # Parameters
    ///
    /// - `reader`: The async readable source
    /// - `capacity`: Buffer size in bytes
    ///
    /// # Examples
    ///
    /// ```rust
    /// # #[cfg(feature = "async")]
    /// use hedl_stream::AsyncLineReader;
    /// use std::io::Cursor;
    ///
    /// // Use a larger buffer for large files
    /// let reader = AsyncLineReader::with_capacity(
    ///     Cursor::new("data"),
    ///     256 * 1024  // 256KB
    /// );
    /// ```
    pub fn with_capacity(reader: R, capacity: usize) -> Self {
        Self {
            reader: BufReader::with_capacity(capacity, reader),
            line_number: 0,
            buffer: String::new(),
            peeked: None,
        }
    }

    /// Get the current line number.
    ///
    /// Returns 0 before any lines are read, then increments with each line.
    ///
    /// # Examples
    ///
    /// ```rust
    /// # #[cfg(feature = "async")]
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// use hedl_stream::AsyncLineReader;
    /// use std::io::Cursor;
    ///
    /// let mut reader = AsyncLineReader::new(Cursor::new("line1\nline2"));
    ///
    /// assert_eq!(reader.line_number(), 0);
    /// reader.next_line().await?;
    /// assert_eq!(reader.line_number(), 1);
    /// reader.next_line().await?;
    /// assert_eq!(reader.line_number(), 2);
    /// # Ok(())
    /// # }
    /// ```
    #[inline]
    pub fn line_number(&self) -> usize {
        self.line_number
    }

    /// Read the next line asynchronously.
    ///
    /// Returns `Ok(Some((line_num, line)))` if a line was read, `Ok(None)` at EOF,
    /// or `Err` on I/O errors.
    ///
    /// Trailing newlines (LF or CRLF) are automatically stripped.
    ///
    /// # Performance
    ///
    /// This method awaits on I/O and yields to the runtime if data is not available,
    /// allowing other tasks to run. It does not block the thread.
    ///
    /// # Examples
    ///
    /// ```rust
    /// # #[cfg(feature = "async")]
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// use hedl_stream::AsyncLineReader;
    /// use std::io::Cursor;
    ///
    /// let mut reader = AsyncLineReader::new(Cursor::new("hello\nworld"));
    ///
    /// let (num, line) = reader.next_line().await?.unwrap();
    /// assert_eq!(num, 1);
    /// assert_eq!(line, "hello");
    ///
    /// let (num, line) = reader.next_line().await?.unwrap();
    /// assert_eq!(num, 2);
    /// assert_eq!(line, "world");
    ///
    /// assert_eq!(reader.next_line().await?, None);
    /// # Ok(())
    /// # }
    /// ```
    pub async fn next_line(&mut self) -> StreamResult<Option<(usize, String)>> {
        // Return peeked line if available
        if let Some(peeked) = self.peeked.take() {
            return Ok(Some(peeked));
        }

        self.buffer.clear();

        match self.reader.read_line(&mut self.buffer).await {
            Ok(0) => Ok(None), // EOF
            Ok(_) => {
                self.line_number += 1;

                // Remove trailing newline
                if self.buffer.ends_with('\n') {
                    self.buffer.pop();
                    if self.buffer.ends_with('\r') {
                        self.buffer.pop();
                    }
                }

                Ok(Some((self.line_number, self.buffer.clone())))
            }
            Err(e) => Err(StreamError::Io(e)),
        }
    }

    /// Peek at the next line without consuming it.
    ///
    /// Returns a reference to the next line without advancing the reader.
    /// Subsequent calls to `peek_line()` return the same line. Call `next_line()`
    /// to consume it.
    ///
    /// # Examples
    ///
    /// ```rust
    /// # #[cfg(feature = "async")]
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// use hedl_stream::AsyncLineReader;
    /// use std::io::Cursor;
    ///
    /// let mut reader = AsyncLineReader::new(Cursor::new("line1\nline2"));
    ///
    /// // Peek multiple times
    /// assert_eq!(reader.peek_line().await?, Some(&(1, "line1".to_string())));
    /// assert_eq!(reader.peek_line().await?, Some(&(1, "line1".to_string())));
    ///
    /// // Consume
    /// reader.next_line().await?;
    ///
    /// // Next peek is the second line
    /// assert_eq!(reader.peek_line().await?, Some(&(2, "line2".to_string())));
    /// # Ok(())
    /// # }
    /// ```
    pub async fn peek_line(&mut self) -> StreamResult<Option<&(usize, String)>> {
        if self.peeked.is_none() {
            self.peeked = self.read_line_internal().await?;
        }
        Ok(self.peeked.as_ref())
    }

    /// Push a line back to be read again.
    ///
    /// The next call to `next_line()` or `peek_line()` will return this line.
    /// Only one line can be pushed back at a time; subsequent calls overwrite
    /// the previously pushed line.
    ///
    /// # Examples
    ///
    /// ```rust
    /// # #[cfg(feature = "async")]
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// use hedl_stream::AsyncLineReader;
    /// use std::io::Cursor;
    ///
    /// let mut reader = AsyncLineReader::new(Cursor::new("line1\nline2"));
    ///
    /// let line = reader.next_line().await?.unwrap();
    /// assert_eq!(line, (1, "line1".to_string()));
    ///
    /// // Push it back
    /// reader.push_back(line.0, line.1);
    ///
    /// // Read it again
    /// let line = reader.next_line().await?.unwrap();
    /// assert_eq!(line, (1, "line1".to_string()));
    /// # Ok(())
    /// # }
    /// ```
    #[inline]
    pub fn push_back(&mut self, line_num: usize, line: String) {
        self.peeked = Some((line_num, line));
    }

    async fn read_line_internal(&mut self) -> StreamResult<Option<(usize, String)>> {
        self.buffer.clear();

        match self.reader.read_line(&mut self.buffer).await {
            Ok(0) => Ok(None),
            Ok(_) => {
                self.line_number += 1;

                if self.buffer.ends_with('\n') {
                    self.buffer.pop();
                    if self.buffer.ends_with('\r') {
                        self.buffer.pop();
                    }
                }

                Ok(Some((self.line_number, self.buffer.clone())))
            }
            Err(e) => Err(StreamError::Io(e)),
        }
    }
}

#[cfg(all(test, feature = "async"))]
mod tests {
    use super::*;
    use std::io::Cursor;

    #[tokio::test]
    async fn test_read_lines() {
        let input = "line1\nline2\nline3";
        let mut reader = AsyncLineReader::new(Cursor::new(input));

        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((1, "line1".to_string()))
        );
        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((2, "line2".to_string()))
        );
        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((3, "line3".to_string()))
        );
        assert_eq!(reader.next_line().await.unwrap(), None);
    }

    #[tokio::test]
    async fn test_peek_and_push_back() {
        let input = "line1\nline2";
        let mut reader = AsyncLineReader::new(Cursor::new(input));

        let peeked = reader.peek_line().await.unwrap().cloned();
        assert_eq!(peeked, Some((1, "line1".to_string())));

        // Should still return the same line
        let line = reader.next_line().await.unwrap();
        assert_eq!(line, Some((1, "line1".to_string())));

        // Push back
        reader.push_back(1, "line1".to_string());
        let line = reader.next_line().await.unwrap();
        assert_eq!(line, Some((1, "line1".to_string())));
    }

    #[tokio::test]
    async fn test_empty_input() {
        let input = "";
        let mut reader = AsyncLineReader::new(Cursor::new(input));
        assert_eq!(reader.next_line().await.unwrap(), None);
    }

    #[tokio::test]
    async fn test_single_empty_line() {
        let input = "\n";
        let mut reader = AsyncLineReader::new(Cursor::new(input));
        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((1, "".to_string()))
        );
        assert_eq!(reader.next_line().await.unwrap(), None);
    }

    #[tokio::test]
    async fn test_crlf_line_endings() {
        let input = "line1\r\nline2\r\nline3";
        let mut reader = AsyncLineReader::new(Cursor::new(input));
        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((1, "line1".to_string()))
        );
        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((2, "line2".to_string()))
        );
        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((3, "line3".to_string()))
        );
    }

    #[tokio::test]
    async fn test_mixed_line_endings() {
        let input = "line1\nline2\r\nline3\nline4";
        let mut reader = AsyncLineReader::new(Cursor::new(input));
        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((1, "line1".to_string()))
        );
        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((2, "line2".to_string()))
        );
        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((3, "line3".to_string()))
        );
        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((4, "line4".to_string()))
        );
    }

    #[tokio::test]
    async fn test_line_number_tracking() {
        let input = "line1\nline2\nline3";
        let mut reader = AsyncLineReader::new(Cursor::new(input));

        assert_eq!(reader.line_number(), 0);

        reader.next_line().await.unwrap();
        assert_eq!(reader.line_number(), 1);

        reader.next_line().await.unwrap();
        assert_eq!(reader.line_number(), 2);

        reader.next_line().await.unwrap();
        assert_eq!(reader.line_number(), 3);
    }

    #[tokio::test]
    async fn test_peek_multiple_times() {
        let input = "line1\nline2";
        let mut reader = AsyncLineReader::new(Cursor::new(input));

        // Peek multiple times should return the same line
        assert_eq!(
            reader.peek_line().await.unwrap(),
            Some(&(1, "line1".to_string()))
        );
        assert_eq!(
            reader.peek_line().await.unwrap(),
            Some(&(1, "line1".to_string()))
        );
        assert_eq!(
            reader.peek_line().await.unwrap(),
            Some(&(1, "line1".to_string()))
        );

        // Consume it
        reader.next_line().await.unwrap();

        // Next peek should be the second line
        assert_eq!(
            reader.peek_line().await.unwrap(),
            Some(&(2, "line2".to_string()))
        );
    }

    #[tokio::test]
    async fn test_with_capacity() {
        let input = "line1\nline2";
        let mut reader = AsyncLineReader::with_capacity(Cursor::new(input), 1024);

        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((1, "line1".to_string()))
        );
        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((2, "line2".to_string()))
        );
    }

    #[tokio::test]
    async fn test_unicode_content() {
        let input = "你好\n世界\n🎉";
        let mut reader = AsyncLineReader::new(Cursor::new(input));

        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((1, "你好".to_string()))
        );
        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((2, "世界".to_string()))
        );
        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((3, "🎉".to_string()))
        );
    }

    #[tokio::test]
    async fn test_long_line() {
        let long_line = "a".repeat(10000);
        let mut reader = AsyncLineReader::new(Cursor::new(long_line.clone()));
        assert_eq!(
            reader.next_line().await.unwrap(),
            Some((1, long_line))
        );
    }

    #[tokio::test]
    async fn test_many_lines() {
        let lines: Vec<String> = (0..1000).map(|i| format!("line{}", i)).collect();
        let input = lines.join("\n");
        let mut reader = AsyncLineReader::new(Cursor::new(input));

        for (i, expected) in lines.iter().enumerate() {
            let result = reader.next_line().await.unwrap();
            assert_eq!(result, Some((i + 1, expected.clone())));
        }
        assert_eq!(reader.next_line().await.unwrap(), None);
    }

    #[tokio::test]
    async fn test_push_back_overwrites_peek() {
        let input = "line1\nline2";
        let mut reader = AsyncLineReader::new(Cursor::new(input));

        reader.peek_line().await.unwrap(); // Peek line1
        reader.push_back(42, "pushed".to_string());

        let line = reader.next_line().await.unwrap();
        assert_eq!(line, Some((42, "pushed".to_string())));
    }
}
