use ccutils::{futures::ArcMutex, sync::IntoArcRwLock};
use futures::FutureExt;
use rune::Any;
use tokio::{io::AsyncBufReadExt, process};

use crate::prelude::*;

#[derive(Debug, Clone)]
struct CommandInfo {
    name: String,
    #[allow(dead_code)]
    args: Vec<String>,
    here: Here,
}

#[derive(Debug, Default)]
struct StreamObserverInner {
    outputs: Vec<String>,
    contains: Vec<(String, Here)>,
}

#[derive(Debug, Clone, Any)]
#[rune(item = ::stories)]
pub(crate) struct StreamObserver {
    inner: ccutils::sync::ArcRwLock<StreamObserverInner>,
}

impl StreamObserver {
    #[rune::function]
    fn contains(&self, text: String, here: Here) -> RResult<()> {
        self.inner
            .write()
            .map_err(|e| anyhow::Error::msg(format!("{:?}", e)))?
            .contains
            .push((text, here));
        Ok(())
    }
}

impl StreamObserver {
    fn finalise(&self, reporter: &Reporter, command_info: &CommandInfo) {
        let inner = self.inner.read().unwrap();
        for contain in inner.contains.iter() {
            if !inner.outputs.iter().any(|x| x.contains(&contain.0)) {
                reporter.report_error(
                    format!(
                        "Command '{}' started at '{}' did not output '{}'.",
                        command_info.name, command_info.here, contain.0
                    ),
                    contain.1.clone(),
                );
            }
        }
    }
}

struct CommandActivity {
    inner: CommandInner,
}

impl Activity for CommandActivity {
    fn finalise(
        &self,
        reporter: &Reporter,
        timeout: std::time::Duration,
    ) -> futures::future::BoxFuture<'static, ()> {
        let inner = self.inner.clone();
        let reporter = reporter.clone();
        async move {
            let mut child = inner.child.lock().await;
            let r = child.try_wait();
            match r {
                Ok(Some(_)) => {}
                Ok(None) => {
                    let r = tokio::time::timeout(timeout, child.wait()).await;

                    // Check if the wait was successful, or need to kill the process
                    match r {
                        Ok(Ok(_)) => {}
                        Ok(Err(_)) | Err(_) => {
                            reporter.handle_result(
                                child.kill().await,
                                Severity::Error,
                                "Failed to stop ",
                                inner.command_info.here.clone(),
                            );
                            if let Err(e) = tokio::time::timeout(timeout, child.wait()).await {
                                reporter.report_error(
                                    format!(
                                        "Failed to wait after stopping '{}', with error '{}'.",
                                        inner.command_info.name, e
                                    ),
                                    inner.command_info.here.clone(),
                                );
                            }
                        }
                    }
                }
                Err(_) => todo!("report error"),
            }

            // Finalise stdout
            inner.stdout.finalise(&reporter, &inner.command_info);
        }
        .boxed()
    }
}

#[derive(Debug, Clone)]
struct CommandInner {
    child: ArcMutex<process::Child>,
    stdout: StreamObserver,
    command_info: CommandInfo,
}

/// Represent an executable command.
#[derive(Debug, Any)]
#[rune(item = ::stories)]
pub(crate) struct Command {
    inner: CommandInner,
}

impl Command {
    /// Create a new context
    pub(crate) fn new(
        context: &Context,
        command_name: String,
        args: Vec<String>,
        here: Here,
    ) -> Result<Command> {
        let mut cmd = tokio::process::Command::new(command_name.clone());
        use std::process::Stdio;
        cmd.args(&args)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped());
        let mut child = cmd.spawn()?;

        let stdout_inner = StreamObserverInner::default().into_arc_rw_lock();

        let stdout = StreamObserver {
            inner: stdout_inner,
        };
        let stdout_channel = child
            .stdout
            .take()
            .ok_or_else(|| anyhow::Error::msg("missing stdout"))?;
        {
            let stdout = stdout.clone();
            context.runtime_ref().spawn(async move {
                let stdout_buffer_reader = tokio::io::BufReader::new(stdout_channel);
                let mut stdout_buffer_reader_lines = stdout_buffer_reader.lines();
                loop {
                    let line = stdout_buffer_reader_lines.next_line().await;
                    match line {
                        Ok(Some(line)) => {
                            stdout.inner.write().unwrap().outputs.push(line);
                        }
                        Ok(None) => {
                            // None means no message was received
                            return;
                        }
                        Err(e) => {
                            log::debug!("stdout/stderr: {}", e);
                            break;
                        }
                    }
                }
            });
        }
        let child = ccutils::futures::arc_mutex_new(child);
        let command_info = CommandInfo {
            name: command_name,
            args,
            here,
        };
        let inner = CommandInner {
            child,
            stdout,
            command_info,
        };
        context
            .inner
            .write()
            .unwrap()
            .push_activity(CommandActivity {
                inner: inner.clone(),
            });

        Ok(Self { inner })
    }
    #[rune::function]
    fn stdout(&self) -> StreamObserver {
        self.inner.stdout.clone()
    }
}
