use std::collections::HashMap;
use std::process::Command;

/// Well-known default ports for common services.
fn well_known_port(image: &str) -> Option<u16> {
    let lower = image.to_lowercase();
    if lower.contains("postgres") {
        Some(5432)
    } else if lower.contains("mysql") || lower.contains("mariadb") {
        Some(3306)
    } else if lower.contains("redis") {
        Some(6379)
    } else if lower.contains("memcached") {
        Some(11211)
    } else if lower.contains("mongo") {
        Some(27017)
    } else if lower.contains("elasticsearch") {
        Some(9200)
    } else if lower.contains("rabbitmq") {
        Some(5672)
    } else if lower.contains("minio") {
        Some(9000)
    } else {
        None
    }
}

/// Health check command for known service images.
fn health_check_cmd(image: &str, container: &str) -> Option<Vec<String>> {
    let lower = image.to_lowercase();
    if lower.contains("postgres") {
        Some(vec![
            "docker".into(),
            "exec".into(),
            container.into(),
            "pg_isready".into(),
            "-q".into(),
        ])
    } else if lower.contains("redis") {
        Some(vec![
            "docker".into(),
            "exec".into(),
            container.into(),
            "redis-cli".into(),
            "ping".into(),
        ])
    } else if lower.contains("mysql") || lower.contains("mariadb") {
        Some(vec![
            "docker".into(),
            "exec".into(),
            container.into(),
            "mysqladmin".into(),
            "ping".into(),
            "--silent".into(),
        ])
    } else if lower.contains("mongo") {
        Some(vec![
            "docker".into(),
            "exec".into(),
            container.into(),
            "mongosh".into(),
            "--eval".into(),
            "db.runCommand('ping')".into(),
        ])
    } else {
        None
    }
}

pub struct ServiceContext {
    pub containers: Vec<String>,
    pub network: String,
    pub env_vars: HashMap<String, String>,
}

/// Start service containers for a job.
/// `services` is a map of service_name → {image, env, port/ports}.
pub async fn start_services(
    job_id: &str,
    services: &serde_json::Value,
) -> Result<ServiceContext, Box<dyn std::error::Error + Send + Sync>> {
    let services_map = services
        .as_object()
        .ok_or("services must be a JSON object")?;

    if services_map.is_empty() {
        return Ok(ServiceContext {
            containers: vec![],
            network: String::new(),
            env_vars: HashMap::new(),
        });
    }

    let network = format!("anvil-ci-{job_id}");

    // Create Docker network
    let output = Command::new("docker")
        .args(["network", "create", &network])
        .output()?;
    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        return Err(format!("failed to create Docker network: {stderr}").into());
    }

    let mut containers = Vec::new();
    let mut env_vars = HashMap::new();

    for (svc_name, svc_config) in services_map {
        let image = svc_config
            .get("image")
            .and_then(|v| v.as_str())
            .ok_or_else(|| format!("service '{svc_name}' missing 'image'"))?;
        let container_name = format!("anvil-ci-svc-{job_id}-{svc_name}");
        let args = build_run_args(svc_name, svc_config, &container_name, &network)
            .map_err(|e| format!("service '{svc_name}': {e}"))?;

        let output = Command::new("docker").args(&args).output()?;
        if !output.status.success() {
            let stderr = String::from_utf8_lossy(&output.stderr);
            // Clean up on failure
            cleanup_services(&containers, &network);
            return Err(format!("failed to start service '{svc_name}': {stderr}").into());
        }

        containers.push(container_name.clone());

        // Wait for health check (up to 30s)
        if let Some(check_cmd) = health_check_cmd(image, &container_name) {
            wait_for_health(&check_cmd, 30).await;
        } else {
            // Brief pause for unknown services
            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
        }

        // Determine port
        let port = svc_config
            .get("port")
            .and_then(|v| v.as_u64())
            .map(|p| p as u16)
            .or_else(|| {
                svc_config
                    .get("ports")
                    .and_then(|v| v.as_array())
                    .and_then(|a| a.first())
                    .and_then(|p| p.as_str())
                    .and_then(|s| s.split(':').next())
                    .and_then(|s| s.parse::<u16>().ok())
            })
            .or_else(|| well_known_port(image));

        // Set env vars for the job
        let svc_upper = svc_name.to_uppercase();
        env_vars.insert(format!("{svc_upper}_HOST"), svc_name.clone());
        if let Some(p) = port {
            env_vars.insert(format!("{svc_upper}_PORT"), p.to_string());
        }

        // Forward service env vars to the job
        if let Some(env_map) = svc_config.get("env").and_then(|v| v.as_object()) {
            for (k, v) in env_map {
                let val = v.as_str().unwrap_or("");
                env_vars.insert(k.clone(), val.to_string());
            }
        }
    }

    Ok(ServiceContext {
        containers,
        network,
        env_vars,
    })
}

/// Wait for a health check command to succeed.
async fn wait_for_health(cmd: &[String], timeout_secs: u64) {
    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(timeout_secs);

    while tokio::time::Instant::now() < deadline {
        if let Ok(output) = Command::new(&cmd[0]).args(&cmd[1..]).output() {
            if output.status.success() {
                return;
            }
        }
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    }
    eprintln!("warning: health check timed out after {timeout_secs}s");
}

/// Build the `docker run` argument list for one service. Pure function —
/// extracted from `start_services` so the arg-building logic (especially
/// the `command:` field that wires through to the container's CMD) is
/// unit-testable without spinning up containers.
pub fn build_run_args(
    svc_name: &str,
    svc_config: &serde_json::Value,
    container_name: &str,
    network: &str,
) -> Result<Vec<String>, String> {
    let image = svc_config
        .get("image")
        .and_then(|v| v.as_str())
        .ok_or_else(|| "missing 'image'".to_string())?;

    let mut args = vec![
        "run".to_string(),
        "--detach".into(),
        "--name".into(),
        container_name.to_string(),
        "--network".into(),
        network.to_string(),
        "--network-alias".into(),
        svc_name.to_string(),
    ];

    if let Some(env_map) = svc_config.get("env").and_then(|v| v.as_object()) {
        for (k, v) in env_map {
            let val = v.as_str().unwrap_or("");
            args.push("-e".into());
            args.push(format!("{k}={val}"));
        }
    }

    if let Some(ports) = svc_config.get("ports").and_then(|v| v.as_array()) {
        for port in ports {
            if let Some(p) = port.as_str() {
                args.push("-p".into());
                args.push(p.to_string());
            }
        }
    }

    args.push(image.to_string());

    // Optional `command:` override (e.g. `server /data` for the official
    // minio/minio image, which has no default CMD that does anything
    // useful — without this it prints help and exits, killing the
    // service before the test container can reach it). Accept either a
    // string (split on whitespace) or a list of pre-split args.
    if let Some(cmd) = svc_config.get("command") {
        if let Some(s) = cmd.as_str() {
            for part in s.split_whitespace() {
                args.push(part.to_string());
            }
        } else if let Some(arr) = cmd.as_array() {
            for part in arr {
                if let Some(s) = part.as_str() {
                    args.push(s.to_string());
                }
            }
        }
    }

    Ok(args)
}

/// Stop and remove all service containers and the network.
pub fn cleanup_services(containers: &[String], network: &str) {
    for container in containers {
        let _ = Command::new("docker")
            .args(["rm", "-f", container])
            .output();
    }
    if !network.is_empty() {
        let _ = Command::new("docker")
            .args(["network", "rm", network])
            .output();
    }
}

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

    #[test]
    fn build_run_args_appends_string_command_after_image() {
        // Regression: minio/minio:latest needs `server /data` as its CMD
        // — without this, the previous code launched it with no args, the
        // entrypoint printed help, and the container exited (code 0)
        // before the test container could reach it. `command:` MUST land
        // AFTER the image in the docker run argv, never before.
        let svc = serde_json::json!({
            "image": "minio/minio:latest",
            "command": "server /data",
            "env": { "MINIO_ROOT_USER": "minioadmin" }
        });

        let args = build_run_args("minio", &svc, "anvil-ci-svc-X-minio", "anvil-ci-X").unwrap();

        let img_pos = args.iter().position(|a| a == "minio/minio:latest").unwrap();
        let server_pos = args.iter().position(|a| a == "server").unwrap();
        let data_pos = args.iter().position(|a| a == "/data").unwrap();
        assert!(server_pos > img_pos, "command must come after image");
        assert_eq!(data_pos, server_pos + 1, "command parts must stay in order");
    }

    #[test]
    fn build_run_args_accepts_list_command() {
        let svc = serde_json::json!({
            "image": "busybox:latest",
            "command": ["sh", "-c", "echo hello"]
        });

        let args = build_run_args("svc", &svc, "c", "n").unwrap();

        let img_pos = args.iter().position(|a| a == "busybox:latest").unwrap();
        assert_eq!(args.get(img_pos + 1), Some(&"sh".to_string()));
        assert_eq!(args.get(img_pos + 2), Some(&"-c".to_string()));
        assert_eq!(args.get(img_pos + 3), Some(&"echo hello".to_string()));
    }

    #[test]
    fn build_run_args_omits_command_when_unset() {
        let svc = serde_json::json!({ "image": "postgres:16" });
        let args = build_run_args("postgres", &svc, "c", "n").unwrap();
        // The image must be the LAST positional arg when no command override.
        assert_eq!(args.last(), Some(&"postgres:16".to_string()));
    }

    #[test]
    fn build_run_args_includes_network_alias_for_dns_resolution() {
        let svc = serde_json::json!({ "image": "x" });
        let args = build_run_args("alias-name", &svc, "c", "net").unwrap();
        let alias_pos = args.iter().position(|a| a == "--network-alias").unwrap();
        assert_eq!(args.get(alias_pos + 1), Some(&"alias-name".to_string()));
    }
}