use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::{Arc, Mutex, OnceLock};

use ci_protocol::runner::{JobPayload, StepPayload};
use config::runner::RunnerConfig;
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::Command;
use tracing::Instrument as _;

pub const MAX_STEP_OUTPUT_BYTES: usize = 4 * 1024 * 1024;

#[derive(Debug)]
pub enum JobOutcome {
    Success {
        exit_code: Option<i32>,
    },
    Failure {
        exit_code: Option<i32>,
        error: Option<String>,
    },
    Cancelled,
    TimedOut,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StepSignal {
    Started,
    Finished,
}

static CANCELLED: OnceLock<Mutex<HashMap<uuid::Uuid, ()>>> = OnceLock::new();

fn cancelled_set() -> &'static Mutex<HashMap<uuid::Uuid, ()>> {
    CANCELLED.get_or_init(|| Mutex::new(HashMap::new()))
}

pub fn request_cancel(job_ids: &[uuid::Uuid]) {
    if let Ok(mut set) = cancelled_set().lock() {
        for id in job_ids {
            set.insert(*id, ());
        }
    }
    #[cfg(unix)]
    {
        if let Ok(pids) = running_pids().lock()
            && let Some(pid) = job_ids.iter().find_map(|id| pids.get(id).copied())
        {
            let _ = nix_kill_group(pid);
        }
    }
}

pub fn clear_cancel(job_id: &uuid::Uuid) {
    if let Ok(mut set) = cancelled_set().lock() {
        set.remove(job_id);
    }
}

pub(crate) fn is_cancelled(job_id: &uuid::Uuid) -> bool {
    cancelled_set()
        .lock()
        .map(|set| set.contains_key(job_id))
        .unwrap_or(false)
}

static RUNNING_PIDS: OnceLock<Mutex<HashMap<uuid::Uuid, i32>>> = OnceLock::new();

fn running_pids() -> &'static Mutex<HashMap<uuid::Uuid, i32>> {
    RUNNING_PIDS.get_or_init(|| Mutex::new(HashMap::new()))
}

#[cfg(unix)]
fn nix_kill_group(pid: i32) {
    // TERM the whole process group so shell children die with the step.
    unsafe {
        libc_kill(-pid, 15);
    }
}

#[cfg(unix)]
unsafe fn libc_kill(pid: i32, sig: i32) {
    unsafe extern "C" {
        fn kill(pid: i32, sig: i32) -> i32;
    }
    unsafe {
        kill(pid, sig);
    }
}

#[derive(Clone)]
pub struct StepHooks {
    pub on_status: std::sync::Arc<dyn Fn(i32, StepSignal, Option<i32>) + Send + Sync>,
    pub on_log: std::sync::Arc<dyn Fn(String) + Send + Sync>,
}

#[allow(clippy::too_many_arguments)]
pub async fn execute_job(
    cfg: &RunnerConfig,
    payload: &JobPayload,
    hooks: StepHooks,
    on_artifact: std::sync::Arc<dyn Fn(String, String, i64, Vec<u8>) + Send + Sync>,
    renew: std::sync::Arc<dyn Fn() + Send + Sync>,
) -> JobOutcome {
    let workspace = PathBuf::from(&cfg.workspace_root)
        .join(payload.run_id.to_string())
        .join(payload.job_id.to_string());
    if let Err(error) = std::fs::create_dir_all(&workspace) {
        return JobOutcome::Failure {
            exit_code: None,
            error: Some(format!("creating workspace failed: {error}")),
        };
    }

    let repo_dir = workspace.join("repo");
    if let Err(error) = prepare_workspace(cfg, payload, &workspace, &repo_dir, hooks.clone()).await
    {
        cleanup(cfg, &workspace);
        return JobOutcome::Failure {
            exit_code: None,
            error: Some(error.to_string()),
        };
    }

    if let Err(error) = download_needs_artifacts(payload, &repo_dir, hooks.clone()).await {
        (hooks.on_log)(format!("artifact download failed: {error}\n"));
    }

    let mut exit_code = 0i32;
    let mut failure: Option<String> = None;
    let deadline =
        std::time::Instant::now() + std::time::Duration::from_secs(payload.timeout_secs.max(1));

    for step in &payload.steps {
        if is_cancelled(&payload.job_id) {
            (hooks.on_status)(step.index, StepSignal::Finished, None);
            cleanup(cfg, &workspace);
            return JobOutcome::Cancelled;
        }
        if std::time::Instant::now() >= deadline {
            cleanup(cfg, &workspace);
            return JobOutcome::TimedOut;
        }
        (hooks.on_status)(step.index, StepSignal::Started, None);
        (hooks.on_log)(format!(
            "::group:: step {} · {}\n",
            step.index + 1,
            step.name
        ));
        let step_span = tracing::info_span!("ci.step", index = step.index, name = %step.name);
        let result = run_step(cfg, payload, step, &repo_dir, hooks.clone(), deadline)
            .instrument(step_span)
            .await;
        match result {
            Ok(code) => {
                (hooks.on_status)(step.index, StepSignal::Finished, Some(code));
                (hooks.on_log)("::endgroup::\n".to_string());
                if code != 0 {
                    exit_code = code;
                    failure = Some(format!("step `{}` exited with {code}", step.name));
                    break;
                }
            }
            Err(StepFailure::Cancelled) => {
                (hooks.on_status)(step.index, StepSignal::Finished, None);
                cleanup(cfg, &workspace);
                return JobOutcome::Cancelled;
            }
            Err(StepFailure::TimedOut) => {
                (hooks.on_status)(step.index, StepSignal::Finished, None);
                cleanup(cfg, &workspace);
                return JobOutcome::TimedOut;
            }
            Err(StepFailure::Io(message)) => {
                (hooks.on_status)(step.index, StepSignal::Finished, None);
                (hooks.on_log)(format!("{message}\n"));
                exit_code = 1;
                failure = Some(message);
                break;
            }
        }
        (renew)();
    }

    collect_artifacts(cfg, payload, &repo_dir, Arc::clone(&on_artifact)).await;
    cleanup(cfg, &workspace);

    match failure {
        None => JobOutcome::Success {
            exit_code: Some(exit_code),
        },
        Some(error) => JobOutcome::Failure {
            exit_code: Some(exit_code),
            error: Some(error),
        },
    }
}

enum StepFailure {
    Cancelled,
    TimedOut,
    Io(String),
}

async fn prepare_workspace(
    cfg: &RunnerConfig,
    payload: &JobPayload,
    workspace: &Path,
    repo_dir: &Path,
    hooks: StepHooks,
) -> anyhow::Result<()> {
    (hooks.on_log)(format!(
        "clone {} @ {}\n",
        payload.clone_url, payload.commit_sha
    ));
    if repo_dir.exists() {
        std::fs::remove_dir_all(repo_dir)?;
    }
    let auth_url = match &cfg.pat {
        pat if !pat.trim().is_empty() => inject_pat(&payload.clone_url, pat),
        _ => payload.clone_url.clone(),
    };
    let mut command = Command::new(&cfg.git_binary);
    command
        .arg("clone")
        .arg("--no-tags")
        .arg(&auth_url)
        .arg(repo_dir)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .current_dir(workspace);
    run_streaming(command, None, &cfg.clone_timeout_secs, hooks.clone()).await?;
    let mut checkout = Command::new(&cfg.git_binary);
    checkout
        .arg("-C")
        .arg(repo_dir)
        .arg("checkout")
        .arg("--force")
        .arg(&payload.commit_sha)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    run_streaming(checkout, None, &cfg.clone_timeout_secs, hooks.clone()).await?;
    Ok(())
}

fn inject_pat(url: &str, pat: &str) -> String {
    match url.split_once("://") {
        Some((scheme, rest)) => format!("{scheme}://{pat}@{rest}"),
        None => url.to_string(),
    }
}

async fn download_needs_artifacts(
    payload: &JobPayload,
    repo_dir: &Path,
    hooks: StepHooks,
) -> anyhow::Result<()> {
    for artifact in &payload.needs_artifacts {
        let response = reqwest::get(&artifact.download_url).await?;
        if !response.status().is_success() {
            anyhow::bail!("artifact {} returned {}", artifact.name, response.status());
        }
        let bytes = response.bytes().await?;
        if let Some(expected) = &artifact.sha256 {
            let actual = sha256_hex(&bytes);
            if &actual != expected {
                anyhow::bail!("artifact {} checksum mismatch", artifact.name);
            }
        }
        let dir = repo_dir.join(".gitdata-artifacts");
        std::fs::create_dir_all(&dir)?;
        std::fs::write(dir.join(&artifact.name), &bytes)?;
        (hooks.on_log)(format!(
            "downloaded artifact {} ({} bytes) into .gitdata-artifacts/\n",
            artifact.name,
            bytes.len()
        ));
    }
    Ok(())
}

async fn run_step(
    cfg: &RunnerConfig,
    payload: &JobPayload,
    step: &StepPayload,
    repo_dir: &Path,
    hooks: StepHooks,
    job_deadline: std::time::Instant,
) -> Result<i32, StepFailure> {
    let workdir = match &step.working_directory {
        Some(dir) if !dir.trim().is_empty() => repo_dir.join(dir.trim()),
        _ => repo_dir.to_path_buf(),
    };
    if !workdir.starts_with(repo_dir) {
        return Err(StepFailure::Io(
            "working-directory escapes the workspace".to_string(),
        ));
    }
    let shell = step
        .shell
        .clone()
        .unwrap_or_else(|| cfg.default_shell.clone());
    let mut env: HashMap<String, String> = payload.env.clone().into_iter().collect();
    env.extend(step.env.clone());
    // Secrets ride on the job payload and are layered under the workflow/step
    // env so a stored secret cannot be shadowed by a plaintext variable.
    for (key, value) in &payload.secrets {
        env.insert(key.clone(), value.clone());
    }
    if payload.allow_git_token && !cfg.pat.trim().is_empty() {
        env.insert("GITDATA_TOKEN".to_string(), cfg.pat.clone());
    }

    let timeout = step
        .timeout_secs
        .map(|secs| std::time::Duration::from_secs(secs.max(1)))
        .unwrap_or_else(|| {
            job_deadline
                .saturating_duration_since(std::time::Instant::now())
                .max(std::time::Duration::from_secs(1))
        });

    let mut command = if let Some(image) = &payload.container {
        if !cfg.docker_enabled {
            return Err(StepFailure::Io(
                "job requires a container but docker is disabled on this runner".to_string(),
            ));
        }
        let mut cmd = Command::new(&cfg.docker_binary);
        cmd.arg("run")
            .arg("--rm")
            .arg("-v")
            .arg(format!("{}:/workspace", repo_dir.display()))
            .arg("-w")
            .arg(match &step.working_directory {
                Some(dir) if !dir.trim().is_empty() => {
                    format!("/workspace/{}", dir.trim().trim_start_matches('/'))
                }
                _ => "/workspace".to_string(),
            });
        for (key, value) in &env {
            cmd.arg("-e").arg(format!("{key}={value}"));
        }
        cmd.arg(image).arg(&shell).arg("-c").arg(&step.run);
        cmd
    } else {
        let mut cmd = Command::new(&shell);
        cmd.arg("-c").arg(&step.run).current_dir(&workdir);
        for (key, value) in &env {
            cmd.env(key, value);
        }
        cmd
    };

    command
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .stdin(Stdio::null());
    #[cfg(unix)]
    {
        command.process_group(0);
    }

    let mut child = command
        .spawn()
        .map_err(|e| StepFailure::Io(format!("spawning step failed: {e}")))?;
    let pid = child.id().map(|p| p as i32);
    if let Some(pid) = pid
        && let Ok(mut pids) = running_pids().lock()
    {
        pids.insert(payload.job_id, pid);
    }

    let stdout = child.stdout.take();
    let stderr = child.stderr.take();
    let stdout_hook = Arc::clone(&hooks.on_log);
    let stdout_task = stdout.map(|stdout| {
        tokio::spawn(async move {
            let mut lines = BufReader::new(stdout).lines();
            while let Ok(Some(line)) = lines.next_line().await {
                stdout_hook(format!("{line}\n"));
            }
        })
    });
    let stderr_hook = Arc::clone(&hooks.on_log);
    let stderr_task = stderr.map(|stderr| {
        tokio::spawn(async move {
            let mut lines = BufReader::new(stderr).lines();
            while let Ok(Some(line)) = lines.next_line().await {
                stderr_hook(format!("{line}\n"));
            }
        })
    });
    // Race the wait against a cancel poll so `docker run` / host steps stop
    // promptly on every platform, not only where SIGTERM reaches a group.
    let wait: Result<Result<std::process::ExitStatus, std::io::Error>, StepFailure> = tokio::select! {
        result = tokio::time::timeout(timeout, child.wait()) => result.map_err(|_| StepFailure::TimedOut),
        _ = async {
            while !is_cancelled(&payload.job_id) {
                tokio::time::sleep(std::time::Duration::from_millis(250)).await;
            }
        } => {
            let _ = child.kill().await;
            let _ = child.wait().await;
            Err(StepFailure::Cancelled)
        }
    };
    if let Some(task) = stdout_task {
        let _ = task.await;
    }
    if let Some(task) = stderr_task {
        let _ = task.await;
    }
    if let Some(_pid) = pid
        && let Ok(mut pids) = running_pids().lock()
    {
        pids.remove(&payload.job_id);
    }

    match wait {
        Ok(Ok(status)) => {
            if is_cancelled(&payload.job_id) {
                return Err(StepFailure::Cancelled);
            }
            Ok(status.code().unwrap_or(-1))
        }
        Ok(Err(error)) => Err(StepFailure::Io(format!("step wait failed: {error}"))),
        Err(StepFailure::Cancelled) => {
            let _ = child.kill().await;
            Err(StepFailure::Cancelled)
        }
        Err(StepFailure::TimedOut) => {
            let _ = child.kill().await;
            if is_cancelled(&payload.job_id) {
                Err(StepFailure::Cancelled)
            } else {
                Err(StepFailure::TimedOut)
            }
        }
        Err(StepFailure::Io(message)) => Err(StepFailure::Io(message)),
    }
}

async fn run_streaming(
    mut command: Command,
    stdin: Option<Vec<u8>>,
    timeout_secs: &u64,
    hooks: StepHooks,
) -> anyhow::Result<()> {
    let _ = stdin;
    let mut child = command.spawn()?;
    let stdout = child.stdout.take();
    let stderr = child.stderr.take();
    let stdout_hook = Arc::clone(&hooks.on_log);
    let stdout_task = stdout.map(|stdout| {
        tokio::spawn(async move {
            let mut lines = BufReader::new(stdout).lines();
            while let Ok(Some(line)) = lines.next_line().await {
                stdout_hook(format!("{line}\n"));
            }
        })
    });
    let stderr_hook = Arc::clone(&hooks.on_log);
    let stderr_task = stderr.map(|stderr| {
        tokio::spawn(async move {
            let mut lines = BufReader::new(stderr).lines();
            while let Ok(Some(line)) = lines.next_line().await {
                stderr_hook(format!("{line}\n"));
            }
        })
    });
    let status = tokio::time::timeout(
        std::time::Duration::from_secs((*timeout_secs).max(1)),
        child.wait(),
    )
    .await;
    if let Some(task) = stdout_task {
        let _ = task.await;
    }
    if let Some(task) = stderr_task {
        let _ = task.await;
    }
    match status {
        Ok(status) => {
            let status = status?;
            if status.success() {
                Ok(())
            } else {
                anyhow::bail!("command exited with {:?}", status.code())
            }
        }
        Err(_) => {
            let _ = child.kill().await;
            anyhow::bail!("command timed out")
        }
    }
}

async fn collect_artifacts(
    _cfg: &RunnerConfig,
    payload: &JobPayload,
    repo_dir: &Path,
    on_artifact: std::sync::Arc<dyn Fn(String, String, i64, Vec<u8>) + Send + Sync>,
) {
    for spec in &payload.artifacts {
        let base = repo_dir.join(&spec.path);
        let mut files: Vec<(String, Vec<u8>)> = Vec::new();
        if base.is_file() {
            if let Ok(bytes) = std::fs::read(&base) {
                files.push((spec.path.clone(), bytes));
            }
        } else if base.is_dir() {
            collect_dir(&base, &spec.path, &mut files);
        } else if let Ok(set) = globset::Glob::new(&spec.path) {
            let matcher = set.compile_matcher();
            walk(repo_dir, repo_dir, &mut |rel, path| {
                if matcher.is_match(rel)
                    && let Ok(bytes) = std::fs::read(path)
                {
                    files.push((rel.to_string(), bytes));
                }
            });
        }
        if files.is_empty() {
            continue;
        }
        // A spec maps to a single artifact; directory/glob payloads are
        // tar-less concatenations preserved per file under the artifact name.
        if files.len() == 1 {
            let (_, bytes) = files.remove(0);
            let size = bytes.len() as i64;
            on_artifact(spec.name.clone(), spec.path.clone(), size, bytes);
        } else {
            let mut bytes = Vec::new();
            for (rel, content) in files {
                bytes.extend_from_slice(format!("--- {rel} ---\n").as_bytes());
                bytes.extend_from_slice(&content);
                bytes.push(b'\n');
            }
            let size = bytes.len() as i64;
            on_artifact(spec.name.clone(), spec.path.clone(), size, bytes);
        }
    }
}

fn collect_dir(dir: &Path, prefix: &str, out: &mut Vec<(String, Vec<u8>)>) {
    walk(dir, dir, &mut |rel, path| {
        if path.is_file()
            && let Ok(bytes) = std::fs::read(path)
        {
            out.push((format!("{prefix}/{rel}"), bytes));
        }
    });
}

fn walk(root: &Path, dir: &Path, visit: &mut dyn FnMut(&str, &Path)) {
    let Ok(entries) = std::fs::read_dir(dir) else {
        return;
    };
    for entry in entries.flatten() {
        let path = entry.path();
        if path.is_dir() {
            walk(root, &path, visit);
        } else if let Ok(rel) = path.strip_prefix(root) {
            visit(&rel.to_string_lossy(), &path);
        }
    }
}

fn cleanup(cfg: &RunnerConfig, workspace: &Path) {
    if cfg.keep_workspace {
        return;
    }
    let _ = std::fs::remove_dir_all(workspace);
}

pub async fn upload_file(url: &str, bytes: Vec<u8>) -> anyhow::Result<()> {
    let span = export::telemetry::http_client_span("PUT", url);
    let response =
        export::telemetry::send(span, reqwest::Client::new().put(url).body(bytes)).await?;
    if !response.status().is_success() {
        anyhow::bail!("upload failed with {}", response.status());
    }
    Ok(())
}

#[must_use]
pub fn sha256_hex(bytes: &[u8]) -> String {
    use sha2::{Digest, Sha256};
    let digest = Sha256::digest(bytes);
    let mut out = String::with_capacity(digest.len() * 2);
    for byte in digest {
        out.push_str(&format!("{byte:02x}"));
    }
    out
}