use std::{
    env,
    ffi::OsString,
    path::{Path, PathBuf},
    process::Stdio,
    sync::Weak,
    time::Duration,
};
use tokio::{io::AsyncWriteExt, process::Command};

const CERTBOT_PACKAGES: &[&str] = &[
    "certbot==5.8.0",
    "certbot-dns-cloudflare==5.8.0",
    "certbot-dns-ovh==5.8.0",
    "certbot-dns-rfc2136==5.8.0",
];

#[derive(Debug, Clone, Serialize)]
pub struct RuntimeStatus {
    pub status: &'static str,
    pub mode: &'static str,
    pub message: &'static str,
    pub managed_directory: Option<String>,
    pub python: bool,
    pub pip: bool,
    pub certbot: bool,
}

impl Default for RuntimeStatus {
    fn default() -> Self {
        Self {
            status: "checking",
            mode: "none",
            message: "acme.runtime_checking",
            managed_directory: None,
            python: false,
            pip: false,
            certbot: false,
        }
    }
}

impl Manager {
    pub fn runtime_view(&self) -> RuntimeStatus {
        self.runtime
            .lock()
            .unwrap_or_else(|error| error.into_inner())
            .clone()
    }

    fn set_runtime(&self, value: RuntimeStatus) {
        *self
            .runtime
            .lock()
            .unwrap_or_else(|error| error.into_inner()) = value;
    }
}

#[derive(Debug)]
struct RuntimeLaunch {
    python: PathBuf,
    prepend_path: Option<PathBuf>,
}

fn find_on_path(name: &str) -> Option<PathBuf> {
    let candidate = Path::new(name);
    if candidate.components().count() > 1 {
        return candidate.is_file().then(|| candidate.to_owned());
    }
    env::var_os("PATH")
        .into_iter()
        .flat_map(|value| env::split_paths(&value).collect::<Vec<_>>())
        .map(|directory| directory.join(name))
        .find(|path| path.is_file())
}

fn system_python() -> Option<PathBuf> {
    find_on_path("python3").or_else(|| find_on_path("python"))
}

#[cfg(windows)]
fn venv_bin(directory: &Path) -> PathBuf {
    directory.join("Scripts")
}
#[cfg(not(windows))]
fn venv_bin(directory: &Path) -> PathBuf {
    directory.join("bin")
}

#[cfg(windows)]
fn venv_python(directory: &Path) -> PathBuf {
    venv_bin(directory).join("python.exe")
}
#[cfg(not(windows))]
fn venv_python(directory: &Path) -> PathBuf {
    venv_bin(directory).join("python")
}

#[cfg(windows)]
fn venv_certbot(directory: &Path) -> PathBuf {
    venv_bin(directory).join("certbot.exe")
}
#[cfg(not(windows))]
fn venv_certbot(directory: &Path) -> PathBuf {
    venv_bin(directory).join("certbot")
}

async fn command_success(
    mut command: Command,
    timeout: Duration,
    shutdown: &tokio_util::sync::CancellationToken,
) -> bool {
    command
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .kill_on_drop(true);
    let Ok(mut child) = command.spawn() else {
        return false;
    };
    tokio::select! {
        _ = shutdown.cancelled() => false,
        result = tokio::time::timeout(timeout, child.wait()) => {
            matches!(result, Ok(Ok(status)) if status.success())
        }
    }
}

async fn managed_environment_valid(
    directory: &Path,
    shutdown: &tokio_util::sync::CancellationToken,
) -> bool {
    let python = venv_python(directory);
    let certbot = venv_certbot(directory);
    if !python.is_file() || !certbot.is_file() {
        return false;
    }
    let mut imports = Command::new(&python);
    imports.args([
        "-I",
        "-c",
        "import certbot, certbot_dns_cloudflare, certbot_dns_ovh, certbot_dns_rfc2136",
    ]);
    if !command_success(imports, Duration::from_secs(20), shutdown).await {
        return false;
    }
    let mut version = Command::new(&certbot);
    version.arg("--version");
    command_success(version, Duration::from_secs(20), shutdown).await
}

async fn system_has_pip(
    python: &Path,
    shutdown: &tokio_util::sync::CancellationToken,
) -> bool {
    let mut command = Command::new(python);
    command.args(["-m", "pip", "--version"]);
    command_success(command, Duration::from_secs(20), shutdown).await
}

async fn create_virtual_environment(
    python: &Path,
    staging: &Path,
    system_pip: bool,
    shutdown: &tokio_util::sync::CancellationToken,
) -> bool {
    let mut create = Command::new(python);
    create.arg("-m").arg("venv").arg(staging);
    if command_success(create, Duration::from_secs(180), shutdown).await {
        return true;
    }
    if !system_pip {
        return false;
    }

    // Some distributions split the stdlib venv module into a separate package.
    // Bootstrap virtualenv with the system pip into the RustProxy data directory
    // instead of modifying site-packages owned by the distribution.
    let bootstrap = staging.with_extension("virtualenv-bootstrap");
    let _ = tokio::fs::remove_dir_all(&bootstrap).await;
    let mut install = Command::new(python);
    install.args([
        "-m",
        "pip",
        "install",
        "--disable-pip-version-check",
        "--no-input",
        "--no-cache-dir",
        "--retries",
        "1",
        "--timeout",
        "15",
        "--target",
    ]);
    install.arg(&bootstrap).arg("virtualenv>=20,<21");
    if !command_success(install, Duration::from_secs(300), shutdown).await {
        let _ = tokio::fs::remove_dir_all(&bootstrap).await;
        return false;
    }
    let mut create = Command::new(python);
    create
        .arg("-c")
        .arg("import runpy,sys; sys.path.insert(0,sys.argv[1]); sys.argv=['virtualenv',sys.argv[2]]; runpy.run_module('virtualenv',run_name='__main__')")
        .arg(&bootstrap)
        .arg(staging);
    let created = command_success(create, Duration::from_secs(180), shutdown).await;
    let _ = tokio::fs::remove_dir_all(&bootstrap).await;
    created
}

#[derive(Clone, Copy, PartialEq, Eq)]
enum BuildFailure {
    Bootstrap,
    Install,
}

async fn build_managed_environment(
    python: &Path,
    target: &Path,
    system_pip: bool,
    shutdown: &tokio_util::sync::CancellationToken,
) -> Result<(), BuildFailure> {
    let staging = target.with_extension("installing");
    let _ = tokio::fs::remove_dir_all(&staging).await;

    if !create_virtual_environment(python, &staging, system_pip, shutdown).await {
        let _ = tokio::fs::remove_dir_all(&staging).await;
        return Err(BuildFailure::Bootstrap);
    }

    let staging_python = venv_python(&staging);
    let mut pip_check = Command::new(&staging_python);
    pip_check.args(["-m", "pip", "--version"]);
    if !command_success(pip_check, Duration::from_secs(30), shutdown).await {
        let mut ensurepip = Command::new(&staging_python);
        ensurepip.args(["-m", "ensurepip", "--upgrade"]);
        if !command_success(ensurepip, Duration::from_secs(120), shutdown).await {
            let _ = tokio::fs::remove_dir_all(&staging).await;
            return Err(BuildFailure::Bootstrap);
        }
    }

    let mut pip = Command::new(&staging_python);
    pip.args([
        "-m",
        "pip",
        "install",
        "--disable-pip-version-check",
        "--no-input",
        "--no-cache-dir",
        "--retries",
        "1",
        "--timeout",
        "15",
    ]);
    pip.args(CERTBOT_PACKAGES.iter().copied());
    if !command_success(pip, Duration::from_secs(300), shutdown).await
        || !managed_environment_valid(&staging, shutdown).await
    {
        let _ = tokio::fs::remove_dir_all(&staging).await;
        return Err(BuildFailure::Install);
    }

    if target.exists() && tokio::fs::remove_dir_all(target).await.is_err() {
        let _ = tokio::fs::remove_dir_all(&staging).await;
        return Err(BuildFailure::Install);
    }
    if tokio::fs::rename(&staging, target).await.is_err() {
        let _ = tokio::fs::remove_dir_all(&staging).await;
        return Err(BuildFailure::Install);
    }
    Ok(())
}

async fn ensure_runtime(state: &App) -> Result<RuntimeLaunch, String> {
    let _permit = state
        .acme
        .runtime_slot
        .clone()
        .acquire_owned()
        .await
        .map_err(|_| "acme.runtime_setup_failed".to_owned())?;
    let Some(data_dir) = state.policy.data_dir.as_deref() else {
        state.acme.set_runtime(RuntimeStatus {
            status: "unavailable",
            mode: "none",
            message: "acme.runtime_data_dir_missing",
            managed_directory: None,
            python: false,
            pip: false,
            certbot: false,
        });
        return Err("acme.runtime_data_dir_missing".into());
    };
    let managed = data_dir.join("acme-python");
    let managed_directory = Some(managed.to_string_lossy().into_owned());
    if managed_environment_valid(&managed, &state.shutdown).await {
        state.acme.set_runtime(RuntimeStatus {
            status: "ready",
            mode: "managed",
            message: "acme.runtime_ready",
            managed_directory,
            python: true,
            pip: true,
            certbot: true,
        });
        return Ok(RuntimeLaunch {
            python: venv_python(&managed),
            prepend_path: Some(venv_bin(&managed)),
        });
    }

    let distro_certbot = find_on_path("certbot");
    let distro_certbot_available = if let Some(certbot) = distro_certbot.as_deref() {
        let mut version = Command::new(certbot);
        version.arg("--version");
        command_success(version, Duration::from_secs(20), &state.shutdown).await
    } else {
        false
    };
    let Some(python) = system_python() else {
        state.acme.set_runtime(RuntimeStatus {
            status: "unavailable",
            mode: "none",
            message: "acme.runtime_python_missing",
            managed_directory,
            python: false,
            pip: false,
            certbot: distro_certbot_available,
        });
        return Err("acme.runtime_python_missing".into());
    };
    let system_pip = system_has_pip(&python, &state.shutdown).await;
    state.acme.set_runtime(RuntimeStatus {
        status: "initializing",
        mode: "managed",
        message: "acme.runtime_initializing",
        managed_directory: managed_directory.clone(),
        python: true,
        pip: system_pip,
        certbot: distro_certbot_available,
    });

    let build = build_managed_environment(&python, &managed, system_pip, &state.shutdown).await;
    if build.is_ok() {
        state.acme.set_runtime(RuntimeStatus {
            status: "ready",
            mode: "managed",
            message: "acme.runtime_ready",
            managed_directory,
            python: true,
            pip: true,
            certbot: true,
        });
        return Ok(RuntimeLaunch {
            python: venv_python(&managed),
            prepend_path: Some(venv_bin(&managed)),
        });
    }

    if distro_certbot_available {
        state.acme.set_runtime(RuntimeStatus {
            status: "ready",
            mode: "system",
            message: "acme.runtime_fallback",
            managed_directory,
            python: true,
            pip: system_pip,
            certbot: true,
        });
        return Ok(RuntimeLaunch {
            python,
            prepend_path: None,
        });
    }

    let message = if build == Err(BuildFailure::Bootstrap) && !system_pip {
        "acme.runtime_pip_missing"
    } else {
        "acme.runtime_setup_failed"
    };
    state.acme.set_runtime(RuntimeStatus {
        status: "unavailable",
        mode: "none",
        message,
        managed_directory,
        python: true,
        pip: system_pip,
        certbot: false,
    });
    Err(message.into())
}

pub async fn runtime_setup_once(state: Weak<App>) {
    let Some(state) = state.upgrade() else {
        return;
    };
    let _ = ensure_runtime(&state).await;
}

/// Kill the entire private process group if execution is cancelled/abandoned.
/// Certbot inherits the group, so it cannot outlive a stopped RustProxy worker.
struct ProcessGroup(Option<u32>);
impl Drop for ProcessGroup {
    fn drop(&mut self) {
        #[cfg(unix)]
        if let Some(id) = self.0 {
            // SAFETY: this is the process group created solely for this child.
            unsafe {
                libc::kill(-(id as i32), libc::SIGKILL);
            }
        }
    }
}

fn merged_path(prepend: &Path) -> Result<OsString, String> {
    let mut values = vec![prepend.to_owned()];
    if let Some(current) = env::var_os("PATH") {
        values.extend(env::split_paths(&current));
    }
    env::join_paths(values).map_err(|_| "acme.runtime_setup_failed".to_owned())
}

async fn worker(state: &App, directory: PathBuf, raw: Zeroizing<String>) -> Result<Value, String> {
    let launch = ensure_runtime(state).await?;
    let mut command = Command::new(&launch.python);
    command
        .arg("-I")
        .arg("-c")
        .arg(include_str!("../../scripts/acme_provider.py"))
        .arg("--cert-dir")
        .arg(&directory)
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::null())
        .kill_on_drop(true);
    if let Some(prepend) = launch.prepend_path.as_deref() {
        command.env("PATH", merged_path(prepend)?);
    }
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt;
        command.as_std_mut().process_group(0);
    }
    let mut child = command
        .spawn()
        .map_err(|_| "acme.python_not_available".to_owned())?;
    let mut group = ProcessGroup(child.id());
    let stdin = child
        .stdin
        .take()
        .ok_or_else(|| "acme.worker_io".to_owned())?;
    let work = async move {
        let mut stdin = stdin;
        stdin
            .write_all(raw.as_bytes())
            .await
            .map_err(|_| "acme.worker_io".to_owned())?;
        stdin
            .shutdown()
            .await
            .map_err(|_| "acme.worker_io".to_owned())?;
        drop(stdin);
        drop(raw);
        child
            .wait_with_output()
            .await
            .map_err(|_| "acme.worker_io".to_owned())
    };
    let output = tokio::select! {
        _ = state.shutdown.cancelled() => return Err("acme.cancelled".into()),
        result = tokio::time::timeout(Duration::from_secs(2160), work) => result.map_err(|_| "acme.provider_timeout".to_owned())??,
    };
    group.0 = None;
    if output.stdout.len() > 16384 {
        return Err("acme.invalid_worker_response".into());
    }
    let value: Value = serde_json::from_slice(&output.stdout)
        .map_err(|_| "acme.invalid_worker_response".to_owned())?;
    if !output.status.success() {
        let reason = value.get("error").and_then(Value::as_str).filter(|value| {
            value.starts_with("acme.")
                && value.len() < 96
                && value
                    .bytes()
                    .all(|byte| byte.is_ascii_lowercase() || byte == b'_' || byte == b'.')
        });
        return Err(reason.unwrap_or("acme.worker_failed").to_owned());
    }
    Ok(value)
}