use std::{
    ffi::OsStr,
    process::{ExitStatus, Stdio},
    sync::Arc,
    time::{Duration, Instant},
};

use pict_rs_error::{ApplicationError, ApplicationResultExt, ErrorCode, Status};
use pict_rs_futures::{
    buf_stream::BufStream,
    future::{LocalBoxFuture, WithTimeout},
    stream::{IntoBufferedStreamer, IntoStreamer, LocalBoxStream, Stream},
};
use tokio::{
    io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt},
    process::{Child, ChildStdin, ChildStdout, Command},
    time::error::Elapsed,
};
use tokio_util::{bytes::Bytes, io::ReaderStream};
use tracing::Instrument;
use uuid::Uuid;

const PROCESS_START: &str = "pict-rs.process.start";
const PROCESS_DURATION: &str = "pict-rs.process.duration";
const PROCESS_END: &str = "pict-rs.process.end";

const COMMAND_ERROR: ErrorCode = ErrorCode::new("command-error");
const COMMAND_TIMEOUT: ErrorCode = ErrorCode::new("command-timeout");
const COMMAND_FAILURE: ErrorCode = ErrorCode::new("command-failure");
const COMMAND_NOT_FOUND: ErrorCode = ErrorCode::new("command-not-found");
const COMMAND_PERMISSION_DENIED: ErrorCode = ErrorCode::new("command-permission-denied");

pub fn describe_metrics() {
    metrics::describe_counter!(
        PROCESS_START,
        "How many times pict-rs has spawned a background process"
    );
    metrics::describe_histogram!(
        PROCESS_DURATION,
        "Timings for how long background processes take to complete"
    );
    metrics::describe_counter!(PROCESS_END, "How many background processes have completed");
}

struct MetricsGuard {
    start: Instant,
    armed: bool,
    command: Arc<str>,
}

impl MetricsGuard {
    fn guard(command: Arc<str>) -> Self {
        metrics::counter!(PROCESS_START, "command" => command.to_string()).increment(1);

        Self {
            start: Instant::now(),
            armed: true,
            command,
        }
    }

    fn disarm(mut self) {
        self.armed = false;
    }
}

impl Drop for MetricsGuard {
    fn drop(&mut self) {
        metrics::histogram!(
            PROCESS_DURATION,
            "command" => self.command.to_string(),
            "completed" => (!self.armed).to_string(),
        )
        .record(self.start.elapsed().as_secs_f64());

        metrics::counter!(PROCESS_END, "completed" => (!self.armed).to_string() , "command" => self.command.to_string()).increment(1);
    }
}

pub struct Process {
    command: Arc<str>,
    child: Child,
    guard: MetricsGuard,
    timeout: Duration,
    extras: Box<dyn Extras>,
    id: Uuid,
}

impl std::fmt::Debug for Process {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Process").field("child", &"Child").finish()
    }
}

#[async_trait::async_trait(?Send)]
pub trait Extras {
    async fn consume(&mut self) -> std::io::Result<()>;
}

#[async_trait::async_trait(?Send)]
impl Extras for () {
    async fn consume(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

#[async_trait::async_trait(?Send)]
impl<T> Extras for (Box<dyn Extras>, T)
where
    T: Extras,
{
    async fn consume(&mut self) -> std::io::Result<()> {
        let (res1, res2) = tokio::join!(self.0.consume(), self.1.consume());
        res1?;
        res2
    }
}

pub struct ProcessRead {
    reader: Reader,
    handle: LocalBoxFuture<'static, Result<(), ApplicationError<ProcessError>>>,
    command: Arc<str>,
    id: Uuid,
    extras: Box<dyn Extras>,
}

#[derive(Debug)]
pub enum ProcessErrorKind {
    NotFound,
    PermissionDenied,
    Timeout,
    Status(ExitStatus),
    Read,
    Cleanup,
    Other,
    SpawnLimit,
    Panic,
}

#[derive(Debug)]
pub struct ProcessError {
    kind: ProcessErrorKind,
    cmd: Arc<str>,
}

impl std::fmt::Display for ProcessError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let cmd = &self.cmd;

        match self.kind {
            ProcessErrorKind::NotFound => write!(
                f,
                "Required command {cmd} not found, make sure it exists in pict-rs' $PATH"
            ),
            ProcessErrorKind::PermissionDenied => write!(
                f,
                "Cannot run command {cmd} due to invalid permissions on binary, make sure the pict-rs user has permission to run it"
            ),
            ProcessErrorKind::Timeout => write!(f, "Command {cmd} exceded time limit"),
            ProcessErrorKind::Status(status) => {
                write!(f, "Command {cmd} failed with status {status}")
            }
            ProcessErrorKind::Read => write!(f, "Failed to read output for command {cmd}"),
            ProcessErrorKind::Cleanup => {
                write!(f, "Failed to cleanup environment for command {cmd}")
            }
            ProcessErrorKind::Other => write!(f, "Unknown process error for command {cmd}"),
            ProcessErrorKind::SpawnLimit => {
                write!(f, "Failed to spawn command {cmd} due to spawn limit")
            }
            ProcessErrorKind::Panic => write!(f, "Failed to spawn command {cmd} due to panic"),
        }
    }
}

impl std::error::Error for ProcessError {}

impl ProcessError {
    fn not_found(cmd: Arc<str>) -> Self {
        Self {
            kind: ProcessErrorKind::NotFound,
            cmd,
        }
    }

    fn permission_denied(cmd: Arc<str>) -> Self {
        Self {
            kind: ProcessErrorKind::PermissionDenied,
            cmd,
        }
    }

    fn timeout(cmd: Arc<str>) -> Self {
        Self {
            kind: ProcessErrorKind::Timeout,
            cmd,
        }
    }

    fn status(cmd: Arc<str>, status: ExitStatus) -> Self {
        Self {
            kind: ProcessErrorKind::Status(status),
            cmd,
        }
    }

    fn read(cmd: Arc<str>) -> Self {
        Self {
            kind: ProcessErrorKind::Read,
            cmd,
        }
    }

    fn cleanup(cmd: Arc<str>) -> Self {
        Self {
            kind: ProcessErrorKind::Cleanup,
            cmd,
        }
    }

    fn other(cmd: Arc<str>) -> Self {
        Self {
            kind: ProcessErrorKind::Other,
            cmd,
        }
    }

    fn limit_reached(cmd: Arc<str>) -> Self {
        Self {
            kind: ProcessErrorKind::SpawnLimit,
            cmd,
        }
    }

    fn panic(cmd: Arc<str>) -> Self {
        Self {
            kind: ProcessErrorKind::Panic,
            cmd,
        }
    }

    pub fn kind(&self) -> &ProcessErrorKind {
        &self.kind
    }
}

async fn handle_finish(
    mut child: Child,
    res: Result<(std::io::Result<ExitStatus>, std::io::Result<()>), Elapsed>,
    guard: MetricsGuard,
    cmd: Arc<str>,
) -> Result<(), ApplicationError<ProcessError>> {
    match res {
        Ok((Ok(status), Ok(()))) if status.success() => {
            guard.disarm();
            Ok(())
        }
        Ok((Ok(status), Err(e))) if status.success() => {
            tracing::warn!("Failed to forward stderr: {e}");
            guard.disarm();
            Ok(())
        }
        Ok((Ok(status), Err(e))) => Err(ApplicationError::new(e)
            .raise(ProcessError::status(cmd, status))
            .error_code(COMMAND_FAILURE)
            .status(Status::Client)),
        Ok((Err(e1), Err(e2))) => Err(ApplicationError::new(e2)
            .raise(e1)
            .error_code(COMMAND_ERROR)
            .raise(ProcessError::other(cmd))),
        Ok((Ok(status), _)) => Err(ApplicationError::new(ProcessError::status(cmd, status))
            .error_code(COMMAND_FAILURE)
            .status(Status::Client)),
        Ok((Err(e), _)) => Err(ApplicationError::new(e)
            .error_code(COMMAND_ERROR)
            .raise(ProcessError::other(cmd))),
        Err(e) => {
            let exn = ApplicationError::new(e);

            match child.kill().with_timeout(Duration::from_secs(1)).await {
                Ok(Ok(())) => Err(exn
                    .error_code(COMMAND_TIMEOUT)
                    .status(Status::Client)
                    .raise(ProcessError::timeout(cmd))),
                Ok(Err(e)) => Err(exn
                    .error_code(COMMAND_TIMEOUT)
                    .status(Status::Client)
                    .raise(e)
                    .raise(ProcessError::timeout(cmd))),
                Err(e) => Err(exn
                    .error_code(COMMAND_TIMEOUT)
                    .status(Status::Client)
                    .raise(e)
                    .raise(ProcessError::timeout(cmd))),
            }
        }
    }
}

#[tracing::instrument(skip_all)]
async fn line_reader<R: tokio::io::AsyncRead + Unpin, F: Fn(&str) + Copy>(
    reader: R,
    f: F,
) -> std::io::Result<()> {
    let mut buf_reader = tokio::io::BufReader::new(reader);
    let mut line = String::new();

    loop {
        let n = buf_reader.read_line(&mut line).await?;

        if n == 0 {
            return Ok(());
        }

        let trimmed = line.trim();
        if !trimmed.is_empty() {
            f(trimmed);
        }

        line.clear();
    }
}

async fn debug_reader<R: tokio::io::AsyncRead + Unpin>(reader: R) -> std::io::Result<()> {
    line_reader(reader, |line| tracing::debug!("{line}")).await
}

impl Process {
    pub async fn run<T>(
        command: &str,
        args: &[T],
        envs: &[(&str, &OsStr)],
        timeout: u64,
    ) -> Result<Self, ApplicationError<ProcessError>>
    where
        T: AsRef<OsStr> + std::fmt::Debug,
    {
        let command: Arc<str> = Arc::from(String::from(command));

        tracing::debug!("{envs:?} {command} {args:?}");

        let mut cmd = Command::new(command.as_ref());
        cmd.args(args).envs(envs.iter().copied());

        Self::spawn(command.clone(), cmd, timeout).await
    }

    async fn spawn(
        command: Arc<str>,
        mut cmd: Command,
        timeout: u64,
    ) -> Result<Self, ApplicationError<ProcessError>> {
        let guard = MetricsGuard::guard(command.clone());

        cmd.stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);

        let res = pict_rs_futures::task::spawn_blocking("spawn-command", move || cmd.spawn())
            .await
            .or_raise(|| ProcessError::panic(command.clone()))?;

        match res {
            Ok(child) => Ok(Process {
                child,
                command,
                guard,
                timeout: Duration::from_secs(timeout),
                extras: Box::new(()),
                id: Uuid::now_v7(),
            }),
            Err(e) => match e.kind() {
                std::io::ErrorKind::NotFound => Err(ApplicationError::new(e)
                    .error_code(COMMAND_NOT_FOUND)
                    .raise(ProcessError::not_found(command))),
                std::io::ErrorKind::PermissionDenied => Err(ApplicationError::new(e)
                    .error_code(COMMAND_PERMISSION_DENIED)
                    .raise(ProcessError::permission_denied(command))),
                std::io::ErrorKind::WouldBlock => Err(ApplicationError::new(e)
                    .error_code(COMMAND_ERROR)
                    .raise(ProcessError::limit_reached(command))),
                _ => Err(ApplicationError::new(e)
                    .error_code(COMMAND_ERROR)
                    .raise(ProcessError::other(command))),
            },
        }
    }

    pub fn add_extras(self, extra: impl Extras + 'static) -> Self {
        Self {
            extras: Box::new((self.extras, extra)),
            ..self
        }
    }

    #[tracing::instrument(skip(self), fields(command = %self.command, id = %self.id))]
    pub async fn wait(self) -> Result<(), ApplicationError<ProcessError>> {
        let Process {
            command,
            mut child,
            guard,
            timeout,
            mut extras,
            id: _,
        } = self;

        let stdin = child.stdin.take().expect("stdin exists");
        drop(stdin);
        let stdout = child.stdout.take().expect("stdout exists");
        let stderr = child.stderr.take().expect("stderr exists");

        let readers = async {
            let (r1, r2) = tokio::join!(debug_reader(stderr), debug_reader(stdout));
            r1?;
            r2
        };

        let res = async { tokio::join!(child.wait(), readers) }
            .with_timeout(timeout)
            .await;

        extras
            .consume()
            .await
            .or_code(COMMAND_ERROR)
            .or_raise(|| ProcessError::cleanup(command.clone()))?;

        handle_finish(child, res, guard, command).await
    }

    pub fn drive_with_stream<S>(self, input: S) -> ProcessRead
    where
        S: Stream<Item = std::io::Result<Bytes>> + 'static,
    {
        self.drive(async move |mut stdin| {
            let stream = std::pin::pin!(input);
            let mut stream = stream.into_buffered_streamer(4);

            while let Some(mut bytes) = stream
                .try_next()
                .with_timeout(Duration::from_secs(5))
                .await??
            {
                match stream
                    .make_progress_with(stdin.write_all_buf(&mut bytes))
                    .await
                {
                    Ok(()) => {}
                    Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe => break,
                    Err(e) => return Err(e),
                }
                pict_rs_futures::task::cooperate().await;
            }

            Ok(())
        })
    }

    pub fn read(self) -> ProcessRead {
        self.drive(async |_| Ok(()))
    }

    #[allow(unknown_lints)]
    #[allow(clippy::let_with_type_underscore)]
    #[tracing::instrument(level = "trace", skip_all)]
    fn drive<F>(self, f: F) -> ProcessRead
    where
        F: AsyncFnOnce(ChildStdin) -> std::io::Result<()> + 'static,
    {
        let Process {
            command,
            mut child,
            guard,
            timeout,
            extras,
            id,
        } = self;

        let stdin = child.stdin.take().expect("stdin exists");
        let stdout = child.stdout.take().expect("stdout exists");
        let stderr = child.stderr.take().expect("stderr exists");

        let command2 = command.clone();
        let handle = Box::pin(async move {
            let child_waiter = async {
                (f)(stdin).await?;

                child.wait().await
            };

            let res = async { tokio::join!(child_waiter, debug_reader(stderr)) }
                .with_timeout(timeout)
                .await;

            handle_finish(child, res, guard, command2).await
        });

        ProcessRead {
            reader: Reader::Stdout(stdout),
            handle,
            command,
            id,
            extras,
        }
    }
}

pub enum Reader {
    Stdout(ChildStdout),
    Stream(LocalBoxStream<'static, std::io::Result<Bytes>>),
}

impl ProcessRead {
    pub fn new<S>(reader: S, command: Arc<str>, id: Uuid) -> Self
    where
        S: Stream<Item = std::io::Result<Bytes>> + 'static,
    {
        Self {
            reader: Reader::Stream(Box::pin(reader)),
            handle: Box::pin(async { Ok(()) }),
            command,
            id,
            extras: Box::new(()),
        }
    }

    pub async fn into_buf_stream(self) -> Result<BufStream, ApplicationError<ProcessError>> {
        let cmd = self.command.clone();

        self.with_reader(async move |reader| match reader {
            Reader::Stdout(stdout) => {
                BufStream::try_from_stream(ReaderStream::with_capacity(stdout, 1024 * 64)).await
            }
            Reader::Stream(stream) => BufStream::try_from_stream(stream).await,
        })
        .await?
        .or_code(COMMAND_ERROR)
        .or_raise(|| ProcessError::read(cmd))
    }

    pub async fn into_vec(self) -> Result<Vec<u8>, ApplicationError<ProcessError>> {
        let cmd = self.command.clone();

        self.with_reader(async move |reader| match reader {
            Reader::Stdout(mut stdout) => {
                let mut vec = Vec::new();

                stdout
                    .read_to_end(&mut vec)
                    .await
                    .or_code(COMMAND_ERROR)
                    .or_raise(|| ProcessError::read(cmd))
                    .map(move |_| vec)
            }
            Reader::Stream(stream) => BufStream::try_from_stream(stream)
                .await
                .or_code(COMMAND_ERROR)
                .or_raise(|| ProcessError::read(cmd))
                .map(BufStream::into_vec),
        })
        .await?
    }

    pub async fn into_string(self) -> Result<String, ApplicationError<ProcessError>> {
        let cmd = self.command.clone();

        self.with_reader(async move |reader| match reader {
            Reader::Stdout(mut stdout) => {
                let mut s = String::new();

                stdout
                    .read_to_string(&mut s)
                    .await
                    .or_code(COMMAND_ERROR)
                    .or_raise(|| ProcessError::read(cmd))
                    .map(move |_| s)
            }
            Reader::Stream(stream) => BufStream::try_from_stream(stream)
                .await
                .or_code(COMMAND_ERROR)
                .or_raise(|| ProcessError::read(cmd.clone()))
                .and_then(|bs| {
                    bs.into_string()
                        .or_code(COMMAND_ERROR)
                        .or_raise(|| ProcessError::read(cmd))
                }),
        })
        .await?
    }

    pub fn pipe(self, process: Process) -> ProcessRead {
        let Process {
            command,
            mut child,
            guard,
            timeout,
            extras,
            id,
        } = process;

        let mut stdin = child.stdin.take().expect("stdin exists");
        let stdout = child.stdout.take().expect("stdout exists");
        let stderr = child.stderr.take().expect("stderr exists");

        let command2 = command.clone();
        let handle = Box::pin(async move {
            self.with_reader(async move |reader| match reader {
                Reader::Stdout(mut stdout) => {
                    let child_waiter = async {
                        tokio::io::copy(&mut stdout, &mut stdin).await?;
                        drop(stdout);
                        drop(stdin);

                        child.wait().await
                    };

                    let res = async { tokio::join!(child_waiter, debug_reader(stderr)) }
                        .with_timeout(timeout)
                        .await;

                    handle_finish(child, res, guard, command2).await
                }
                Reader::Stream(stream) => {
                    let child_waiter = async {
                        let mut streamer = stream.into_streamer();

                        while let Some(mut bytes) = streamer.try_next().await? {
                            stdin.write_all_buf(&mut bytes).await?;
                        }
                        drop(stdin);

                        child.wait().await
                    };

                    let res = async { tokio::join!(child_waiter, debug_reader(stderr)) }
                        .with_timeout(timeout)
                        .await;

                    handle_finish(child, res, guard, command2).await
                }
            })
            .await?
        });

        ProcessRead {
            reader: Reader::Stdout(stdout),
            handle,
            command,
            id,
            extras,
        }
    }

    pub async fn with_reader<F, T>(self, f: F) -> Result<T, ApplicationError<ProcessError>>
    where
        F: AsyncFnOnce(Reader) -> T,
    {
        let Self {
            reader,
            handle,
            command,
            id,
            mut extras,
        } = self;

        let (out, process_result) = tokio::join!(
            (f)(reader).instrument(tracing::info_span!("cmd-reader", %command, %id)),
            handle.instrument(tracing::info_span!("cmd-handle", %command, %id))
        );

        let cleanup_result = extras.consume().await;

        match (process_result, cleanup_result) {
            (Ok(()), Ok(())) => Ok(out),
            (Ok(()), Err(e)) => {
                tracing::warn!("Failed to cleanup afer process, {e}");
                Ok(out)
            }
            (Err(e), Ok(())) => Err(e),
            (Err(e1), Err(e2)) => Err(e1.raise(e2).raise(ProcessError::cleanup(command))),
        }
    }

    pub fn add_extras<E: Extras + 'static>(self, more_extras: E) -> ProcessRead {
        let Self {
            reader,
            handle,
            command,
            id,
            extras,
        } = self;

        Self {
            reader,
            handle,
            command,
            id,
            extras: Box::new((extras, more_extras)),
        }
    }
}
