rostrum 14.0.1

use std::time::Duration;

use anyhow::{bail, Context, Result};

use tokio::{
    io::{AsyncBufReadExt, BufReader},
    sync::mpsc::Sender,
    time::timeout,
};

use crate::signal::Waiter;

use super::Message;

/// Generic request parser for line-based protocols
/// Works with any AsyncBufRead stream (TCP, TLS, etc.)
pub async fn parse_requests<R, F>(
    mut reader: BufReader<R>,
    tx: Sender<Message>,
    idle_timeout: u64,
    line_validator: F,
) -> Result<()>
where
    R: tokio::io::AsyncRead + Unpin,
    F: Fn(&[u8]) -> Result<()>,
{
    loop {
        if Waiter::shutdown_check().is_err() {
            let _ = tx.send(Message::Done).await;
            bail!("shutdown");
        }
        let mut line = Vec::<u8>::new();

        // Setting a timeout for the read_until operation
        let read_result = timeout(
            Duration::from_secs(idle_timeout),
            reader.read_until(b'\n', &mut line),
        )
        .await;

        match read_result {
            Ok(Ok(bytes_read)) => {
                if bytes_read == 0 {
                    tx.send(Message::Done).await.context("channel closed")?;
                    return Ok(());
                }

                // Validate the raw line data (e.g., check for SSL data)
                if let Err(e) = line_validator(&line) {
                    let _ = tx.send(Message::Done).await;
                    return Err(e);
                }

                match String::from_utf8(line) {
                    Ok(req) => tx
                        .send(Message::Request(req))
                        .await
                        .context("channel closed")?,
                    Err(err) => {
                        let _ = tx.send(Message::Done).await;
                        bail!("invalid UTF8: {}", err)
                    }
                }
            }
            // Handling the timeout case
            Err(tokio::time::error::Elapsed { .. }) => {
                let _ = tx.send(Message::Done).await;
                bail!(
                    "idle client did not send a message for {} seconds",
                    idle_timeout
                );
            }
            Ok(Err(e)) => {
                let _ = tx.send(Message::Done).await;
                bail!("failed to read a request: {}", e);
            }
        }
    }
}

/// Default validator that accepts all lines
pub fn no_validation(_line: &[u8]) -> Result<()> {
    Ok(())
}

/// TCP validator that checks for SSL handshake data
pub fn tcp_line_validator(line: &[u8]) -> Result<()> {
    if line.starts_with(&[22, 3, 1]) {
        // (very) naive SSL handshake detection
        bail!("invalid request - maybe SSL-encrypted data?: {:?}", line)
    }
    Ok(())
}