use std::{
    collections::HashMap,
    future::IntoFuture,
    sync::{Arc, Mutex},
};

use colored::Colorize;
use futures::{
    FutureExt,
    channel::oneshot,
    future::{Shared, join_all},
};

use super::runnerstate::RunnerState;
use crate::{
    facts::Facts,
    jobs::Job,
    preflight::RawJob,
    status::{Status, Unsatisfying},
    tags::tags_match,
    template,
};

// TODO: consider extracting the concern of println!ing Status
pub(crate) async fn run(raw_jobs: Vec<RawJob>, facts: &Facts) -> RunnerState {
    let needs: HashMap<String, Vec<String>> = raw_jobs
        .iter()
        .map(|j| (j.name.clone(), j.needs.clone()))
        .collect();

    // unwrap: this is core functionality, so either succeed or panic is fine
    let (renderer, context) = template::make_tera(facts).unwrap();
    let renderer_and_context = Arc::new(Mutex::new((renderer, context)));

    let mut settled_statuses: HashMap<&str, Shared<_>> = HashMap::new();
    let mut status_senders: HashMap<&str, oneshot::Sender<_>> = HashMap::new();
    for name in needs.keys() {
        let (tx, rx) = oneshot::channel::<Status>();
        settled_statuses.insert(name, rx.into_future().shared());
        status_senders.insert(name, tx);
    }

    let mut job_futures: Vec<_> = vec![];
    for mut raw_job in raw_jobs {
        let name = raw_job.name.clone();
        // expect: runner is very broken if a channel is missing
        let tx = status_senders
            .remove(name.as_str())
            .expect("cannot find status sender");
        let renderer_and_context = renderer_and_context.clone();
        let settled_statuses = &settled_statuses;
        job_futures.push(async move {
            // expect: runner is very broken if we somehow cannot receive job status
            let needs_statuses: Vec<Status> =
                join_all(needs_futures(settled_statuses, &raw_job.needs))
                    .await
                    .into_iter()
                    .map(|r| r.expect("broken status channel"))
                    .collect();
            if needs_statuses.iter().any(|s| !s.is_satisfying()) {
                let status = Status::Unsatisfying(Unsatisfying::Blocked);
                // expect: runner is very broken if we somehow cannot send job status
                tx.send(status.clone())
                    .expect("cannot report status of job");
                println!("job: {}: {}", &name, &status);
                return (name, status, None);
            }

            let job = match raw_job.try_into_job(&renderer_and_context) {
                Ok(j) => j,
                Err(e) => {
                    let status = Status::Unsatisfying(Unsatisfying::Error);
                    // expect: runner is very broken if we somehow cannot send job status
                    tx.send(status.clone())
                        .expect("cannot report status of job");
                    eprintln!("{}", format!("job: {}: {:#?}", &name, e).red());
                    return (name, status, None);
                }
            };

            if !tags_match(
                job.metadata.tags.iter().map(String::as_str),
                facts.skip_tags.iter().map(String::as_str),
                facts.tags.iter().map(String::as_str),
            ) {
                let status = Status::Unsatisfying(Unsatisfying::Skipped);
                // expect: runner is very broken if we somehow cannot send job status
                tx.send(status.clone())
                    .expect("cannot report status of job");
                println!("job: {}: {}", &name, &status);
                return (name, status, None);
            }

            let status = run_job(&job).await;
            // expect: runner is very broken if we somehow cannot send job status
            tx.send(status.clone())
                .expect("cannot report status of job");
            (name, status, Some(job))
        });
    }

    let output: Vec<(String, Status, Option<Job>)> = join_all(job_futures).await;

    let mut rs = RunnerState {
        ..Default::default()
    };
    for (name, status, maybe_job) in output {
        if let Some(job) = maybe_job {
            rs.jobs.insert(name.clone(), job);
        }
        rs.statuses.insert(name, status);
    }

    eprintln!("{}", &rs);

    rs
}

fn needs_futures(
    settled_statuses: &HashMap<&str, Shared<oneshot::Receiver<Status>>>,
    needs: &[String],
) -> Vec<Shared<oneshot::Receiver<Status>>> {
    needs
        .iter()
        // expect: missing job indicates runner is too broken to continue
        .map(|n| {
            settled_statuses
                .get(n.as_str())
                .expect("cannot get status channel for job")
                .clone()
        })
        .collect()
}

async fn run_job(job: &Job) -> Status {
    let name = job.metadata.name.clone();

    if !job.when() {
        let status = Status::Unsatisfying(Unsatisfying::Skipped);
        println!("job: {}: {}", &name, &status);
        return status;
    }

    let result = job.execute().await;
    match result {
        Ok(s) => {
            println!("job: {}: {}", &name, s);
            s
        }
        Err(e) => {
            println!("{}", format!("job: {}: {:#?}", &name, e).red());
            Status::Unsatisfying(Unsatisfying::Error)
        }
    }
}

#[cfg(test)]
mod tests {
    use std::{collections::HashMap, time::Duration};

    use toml::Value;

    use crate::{jobs, status::Satisfying};

    use super::*;

    fn assert_executed(state: &RunnerState, name: &str) {
        let history = get_history_by_name(state, name).expect("history not found");
        assert_ne!(history.created, history.updated.get());
    }

    fn assert_not_executed(state: &RunnerState, name: &str) {
        // here we examine the side-effects within the runner's state,
        // but a bug could allow the runner's state to not reflect an actual job execution
        // TODO: find a more bulletproof way to confirm this, e.g. check job's own execution side-effects
        let status = state.statuses.get(name).expect("result not found");
        assert!(status.is_settled());
        assert!(
            status == &Status::Unsatisfying(Unsatisfying::Blocked)
                || status == &Status::Unsatisfying(Unsatisfying::Skipped)
        );
    }

    fn assert_updated_order(state: &RunnerState, earlier: &str, later: &str) {
        let earlier_history = get_history_by_name(state, earlier).expect("history not found");
        let later_history = get_history_by_name(state, later).expect("history not found");
        assert!(earlier_history.updated < later_history.updated);
    }

    fn get_history_by_name(state: &RunnerState, name: &str) -> Option<jobs::History> {
        state.jobs.get(name).map(|j| j.history.clone())
    }

    fn make_fake_job<S>(name: S, status: &Status) -> RawJob
    where
        S: AsRef<str>,
    {
        RawJob {
            name: String::from(name.as_ref()),
            needs: vec![],
            spec: HashMap::from([
                (String::from("status"), Value::String(String::from(status))),
                (String::from("type"), Value::String(String::from("fake"))),
            ]),
        }
    }

    fn make_fake_job_sleep<S>(name: S, sleep: Duration, status: &Status) -> RawJob
    where
        S: AsRef<str>,
    {
        let mut rj = make_fake_job(name, status);
        rj.spec.insert(
            String::from("sleep_ms"),
            Value::Integer(
                sleep
                    .as_millis()
                    .try_into()
                    .expect("cannot convert duration"),
            ),
        );
        rj
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn run_does_not_execute_job_with_false_when_or_needs_job_with_false_when() {
        let mut raw_a = make_fake_job("a", &Status::Satisfying(Satisfying::default()));
        let mut raw_b = make_fake_job("b", &Status::Satisfying(Satisfying::default()));
        raw_a
            .spec
            .insert(String::from("when"), Value::Boolean(false));
        raw_b.needs = vec![String::from("a")];

        let raw_jobs = vec![raw_a, raw_b];
        let facts = Facts::default();
        let state = run(raw_jobs, &facts).await;

        assert_not_executed(&state, "a");
        assert_not_executed(&state, "b");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn run_executes_unordered_jobs() {
        const MAX_COUNT: usize = 10;
        let mut raw_jobs = Vec::<RawJob>::with_capacity(MAX_COUNT);
        for i in 0..MAX_COUNT {
            let job = make_fake_job(
                format!("{i}"),
                &match i % 2 {
                    0 => Status::Satisfying(Satisfying::default()),
                    _ => Status::Satisfying(Satisfying::NoChange(format!("{i}"))),
                },
            );
            raw_jobs.push(job);
        }

        let facts = Facts::default();
        let state = run(raw_jobs, &facts).await;

        for j in state.jobs.values().collect::<Vec<&Job>>() {
            assert_executed(&state, &j.metadata.name);
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn run_executes_unordered_jobs_concurrently() {
        let base_sleep_ms = 250;
        let fake_sleep_ms = base_sleep_ms * 3;
        let raw_a = make_fake_job_sleep(
            "a",
            Duration::from_millis(fake_sleep_ms),
            &Status::Satisfying(Satisfying::default()),
        );
        let raw_b = make_fake_job_sleep(
            "b",
            Duration::from_millis(fake_sleep_ms),
            &Status::Satisfying(Satisfying::default()),
        );

        let raw_jobs = vec![raw_a, raw_b];
        let facts = Facts::default();
        let state = run(raw_jobs, &facts).await;

        assert_executed(&state, "a");
        assert_executed(&state, "b");

        // assert that both jobs finished very recently,
        // that they had to have been executed concurrently
        let history_a = get_history_by_name(&state, "a").expect("history not found");
        let history_b = get_history_by_name(&state, "b").expect("history not found");
        assert!(history_a.created.elapsed() > Duration::from_millis(fake_sleep_ms));
        assert!(history_b.created.elapsed() > Duration::from_millis(fake_sleep_ms));
        assert!(history_a.updated.get().elapsed() < Duration::from_millis(base_sleep_ms + 50));
        assert!(history_b.updated.get().elapsed() < Duration::from_millis(base_sleep_ms + 50));
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn run_executes_jobs_with_complex_needs() {
        const MAX_COUNT: usize = 100;
        let mut raw_jobs = Vec::<RawJob>::with_capacity(MAX_COUNT);
        for i in 0..MAX_COUNT {
            let mut job = make_fake_job(format!("{i}"), &Status::Satisfying(Satisfying::default()));
            match i % 10 {
                2 => {
                    job.needs = vec![format!("{}", i + 2)];
                }
                3 => {
                    job.needs = vec![format!("{}", i - 3)];
                }
                4 => {
                    job.needs = vec![format!("{}", i + 3)];
                }
                7 => {
                    job.needs = vec![String::from("99")];
                }
                _ => { /* noop */ }
            }
            raw_jobs.push(job);
        }

        let facts = Facts::default();
        let state = run(raw_jobs, &facts).await;

        for j in state.jobs.values().collect::<Vec<&Job>>() {
            assert_executed(&state, &j.metadata.name);
            let i = j.metadata.name.parse::<usize>().unwrap_or_else(|err| {
                panic!("could not parse '{}' as usize: {:?}", j.metadata.name, err);
            });
            match i % 10 {
                2 => {
                    // jobs ending in 2 should all run after the next job ending in 4
                    assert_updated_order(&state, &format!("{}", i + 2), &j.metadata.name);
                }
                3 => {
                    // jobs ending in 3 should all run after the previous job ending in 0
                    assert_updated_order(&state, &format!("{}", i - 3), &j.metadata.name);
                }
                4 => {
                    // jobs ending in 4 should all run after the next job ending in 7
                    assert_updated_order(&state, &format!("{}", i + 3), &j.metadata.name);
                }
                7 => {
                    // jobs ending in 7 should all run after job #99
                    assert_updated_order(&state, &format!("{}", 99), &j.metadata.name);
                }
                _ => { /* noop */ }
            }
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn run_executes_ordered_jobs() {
        let mut raw_a = make_fake_job("a", &Status::Satisfying(Satisfying::default()));
        let raw_b = make_fake_job("b", &Status::Satisfying(Satisfying::default()));
        raw_a.needs = vec![String::from("b")];

        let raw_jobs = vec![raw_a, raw_b];
        let facts = Facts::default();
        let state = run(raw_jobs, &facts).await;

        assert_executed(&state, "a");
        assert_executed(&state, "b");
        // assert that "a" finished after "b"
        assert_updated_order(&state, "b", "a");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn run_does_not_execute_ordered_job_when_needs_are_not_done() {
        let mut raw_a = make_fake_job("a", &Status::Satisfying(Satisfying::default()));
        let raw_b = make_fake_job("b", &Status::Unsatisfying(Unsatisfying::Blocked));
        raw_a.needs = vec![String::from("b")];

        let raw_jobs = vec![raw_a, raw_b];
        let facts = Facts::default();
        let state = run(raw_jobs, &facts).await;

        assert_not_executed(&state, "a");
        assert_executed(&state, "b");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn run_does_not_execute_ordered_job_when_some_needs_are_not_done() {
        let mut raw_a = make_fake_job("a", &Status::Satisfying(Satisfying::default()));
        let mut raw_b = make_fake_job("b", &Status::Unsatisfying(Unsatisfying::Blocked));
        let raw_c = make_fake_job("c", &Status::Satisfying(Satisfying::default()));
        raw_a.needs = vec![String::from("b"), String::from("c")];
        raw_b.needs = vec![String::from("c")];

        let raw_jobs = vec![raw_a, raw_b, raw_c];
        let facts = Facts::default();
        let state = run(raw_jobs, &facts).await;

        assert_not_executed(&state, "a");
        assert_executed(&state, "b");
        assert_executed(&state, "c");
    }
}
