//! Local process provider — runs tests as child processes on the local machine.

use std::path::{Path, PathBuf};
use std::process::Stdio;

use async_trait::async_trait;
use futures::stream::{self, StreamExt};
use tokio::io::{AsyncBufReadExt, BufReader};

use super::{
    Command, CostEstimate, OutputLine, OutputStream, ProviderError, ProviderResult, Sandbox,
    SandboxProvider,
};
use crate::config::{LocalProviderConfig, SandboxConfig};

/// Provider that runs tests as local child processes.
///
/// This is the simplest provider implementation. Each sandbox is just
/// a logical grouping with a shared configuration - commands are run
/// as child processes of the offload process itself.
///
/// # Thread Safety
///
/// The provider is thread-safe and can be shared across async tasks.
pub struct LocalProvider {
    config: LocalProviderConfig,
}

impl LocalProvider {
    /// Creates a new process provider with the given configuration.
    ///
    /// # Arguments
    ///
    /// * `config` - Configuration specifying working directory, environment
    ///   variables, and shell to use
    pub fn new(config: LocalProviderConfig) -> Self {
        Self { config }
    }
}

#[async_trait]
impl SandboxProvider for LocalProvider {
    type Sandbox = LocalSandbox;

    async fn prepare(
        &mut self,
        _copy_dirs: &[(PathBuf, PathBuf)],
        _no_cache: bool,
        _sandbox_init_cmd: Option<&str>,
        _discovery_done: Option<&std::sync::atomic::AtomicBool>,
        _context_dir: Option<&std::path::Path>,
    ) -> ProviderResult<Option<String>> {
        // Local provider has no image preparation step.
        Ok(None)
    }

    async fn create_sandbox(&self, config: &SandboxConfig) -> ProviderResult<LocalSandbox> {
        let working_dir = config
            .working_dir
            .as_ref()
            .map(PathBuf::from)
            .or_else(|| self.config.working_dir.clone())
            .unwrap_or_else(|| std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")));

        Ok(LocalSandbox {
            id: config.id.clone(),
            working_dir,
            env: config.env.clone(),
            shell: self.config.shell.clone(),
        })
    }

    fn base_env(&self) -> Vec<(String, String)> {
        self.config
            .env
            .iter()
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect()
    }
}

/// A sandbox that runs commands as local child processes.
///
/// Each command is executed via the configured shell (default: `/bin/sh`).
/// The sandbox provides a consistent working directory and environment
/// for all commands.
///
/// # File Download
///
/// Download operations are implemented as local file copies relative
/// to the working directory. This is useful for tests that produce
/// output files.
///
/// # Termination
///
/// Since processes are transient, termination is a no-op. The sandbox
/// can be safely dropped without cleanup.
pub struct LocalSandbox {
    id: String,
    working_dir: PathBuf,
    env: Vec<(String, String)>,
    shell: String,
}

#[async_trait]
impl Sandbox for LocalSandbox {
    fn id(&self) -> &str {
        &self.id
    }

    async fn exec_stream(&self, cmd: &Command) -> ProviderResult<OutputStream> {
        let shell_cmd = cmd.to_shell_string();

        let mut process = tokio::process::Command::new(&self.shell);
        process.arg("-c").arg(&shell_cmd);
        process.current_dir(&self.working_dir);

        for (key, value) in &self.env {
            process.env(key, value);
        }
        for (key, value) in &cmd.env {
            process.env(key, value);
        }

        if let Some(dir) = &cmd.working_dir {
            process.current_dir(dir);
        }

        process.stdout(Stdio::piped());
        process.stderr(Stdio::piped());

        let mut child = process
            .spawn()
            .map_err(|e| ProviderError::ExecFailed(e.to_string()))?;

        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| ProviderError::ExecFailed("stdout not captured".to_string()))?;
        let stderr = child
            .stderr
            .take()
            .ok_or_else(|| ProviderError::ExecFailed("stderr not captured".to_string()))?;

        let stdout_reader = BufReader::new(stdout);
        let stderr_reader = BufReader::new(stderr);

        let stdout_stream = tokio_stream::wrappers::LinesStream::new(stdout_reader.lines()).map(
            |line: Result<String, std::io::Error>| OutputLine::Stdout(line.unwrap_or_default()),
        );

        let stderr_stream = tokio_stream::wrappers::LinesStream::new(stderr_reader.lines()).map(
            |line: Result<String, std::io::Error>| OutputLine::Stderr(line.unwrap_or_default()),
        );

        // Merge stdout and stderr streams
        let combined = stream::select(stdout_stream, stderr_stream);

        Ok(Box::pin(combined))
    }

    async fn download(&self, paths: &[(&Path, &Path)]) -> ProviderResult<()> {
        for (remote, local) in paths {
            let src = self.working_dir.join(remote);

            if let Some(parent) = local.parent() {
                tokio::fs::create_dir_all(parent)
                    .await
                    .map_err(|e| ProviderError::DownloadFailed(e.to_string()))?;
            }

            if src.is_dir() {
                copy_dir_all(&src, local)
                    .await
                    .map_err(|e| ProviderError::DownloadFailed(e.to_string()))?;
            } else {
                tokio::fs::copy(&src, local)
                    .await
                    .map_err(|e| ProviderError::DownloadFailed(e.to_string()))?;
            }
        }

        Ok(())
    }

    async fn terminate(&self) -> ProviderResult<()> {
        // Process sandboxes don't need explicit cleanup
        Ok(())
    }

    fn cost_estimate(&self) -> CostEstimate {
        CostEstimate::default()
    }
}

/// Recursively copy a directory.
async fn copy_dir_all(src: &Path, dst: &Path) -> std::io::Result<()> {
    tokio::fs::create_dir_all(dst).await?;

    let mut entries = tokio::fs::read_dir(src).await?;
    while let Some(entry) = entries.next_entry().await? {
        let ty = entry.file_type().await?;
        let src_path = entry.path();
        let dst_path = dst.join(entry.file_name());

        if ty.is_dir() {
            Box::pin(copy_dir_all(&src_path, &dst_path)).await?;
        } else {
            tokio::fs::copy(&src_path, &dst_path).await?;
        }
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn local_sandbox_cost_estimate_is_zero() {
        let sandbox = LocalSandbox {
            id: "local-1".to_string(),
            working_dir: PathBuf::from("."),
            env: vec![],
            shell: "/bin/sh".to_string(),
        };

        let cost = sandbox.cost_estimate();
        assert_eq!(cost.cpu_seconds, 0.0);
        assert_eq!(cost.estimated_cost_usd, 0.0);
    }
}
