//! Turns at what runs share: a job, of the `--jobs` at once, and keys
//! each held by one run at a time, such as a workflow's turn on a ref
//! whose older runs finish. Of the jobs, runs other than `Main` hold only
//! so many at once, so a burst of them, as every open pull run again when
//! the branch they are into moves, leaves the rest of the machine room;
//! and each owner's runs only as many as the run waiting may, so one
//! owner's burst leaves room for the rest.
//!
//! A run takes all it needs at once, once all of it is free, so it never
//! holds one thing while it waits for another, and two runs never wait on
//! each other. Runs waiting go by rank: one ranked before another has
//! first claim on the keys it waits for, and on a job when both could
//! start. One ranked lower goes ahead only with keys none before it waits
//! for and a job left over, so a run waiting for a volume in use does not
//! keep others from the jobs free. A run going is never stopped for one
//! ranked before it.
//!
//! Within a class, owners take turns: each of an owner's runs has the
//! turn after its last still waiting, or with none, the turn of the last
//! run to start, so a burst of one owner's runs goes one by one between
//! others' rather than all before them. One owner's runs go in the order
//! they were started.

use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::{Arc, Mutex};
use tokio::sync::Notify;

/// What a run is, for its place in line, first first.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum Class {
    /// A push to its repository's default branch, which deploys wait on.
    Main,
    /// A pull's, or a push to any other ref.
    Other,
    /// A pull's not in the merge queue the runner lands, whose runs go
    /// after the queue's.
    Later,
    /// One its owner asked for by name, as one too costly for every push:
    /// none waits for it.
    Asked,
}

/// A run's class, and the order runs were started in.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub struct Rank {
    pub class: Class,
    pub seq: u64,
}

/// A run's place in line: by its class, then its owner's turn, then
/// oldest first.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
struct Place {
    class: Class,
    turn: u64,
    seq: u64,
}

pub struct Turns {
    board: Mutex<Board>,
    /// Rung whenever what is held, or who waits, changes.
    changed: Notify,
}

/// The jobs free, those of them runs other than `Main` may still take,
/// the keys held and the jobs each owner does, what each run waiting needs, and the turn of the last run to
/// start in each class.
struct Board {
    jobs: usize,
    others: usize,
    held: HashSet<String>,
    owners: HashMap<String, usize>,
    waiting: BTreeMap<Place, Want>,
    last: HashMap<Class, u64>,
}

/// Whose a run is: its repository's owner, and the most jobs that
/// owner's runs may hold for it to start.
pub struct Whose {
    pub owner: String,
    pub most: usize,
}

/// Whose a run waiting is, and the keys it needs.
struct Want {
    whose: Whose,
    keys: Vec<String>,
}

impl Board {
    /// Whether the run at `place` may start now: with those before it
    /// that can start counted as started, and the keys those that cannot
    /// wait for kept for them.
    fn starts(&self, place: &Place, want: &Want) -> bool {
        let mut jobs = (self.jobs, self.others);
        let mut ahead = HashMap::new();
        let mut claimed = HashSet::new();
        for (theirs, w) in self.waiting.range(..place) {
            if self.room(theirs.class, jobs, w, &ahead) && self.free(&w.keys, &claimed) {
                jobs = taken(theirs.class, jobs);
                *ahead.entry(w.whose.owner.as_str()).or_default() += 1;
            }
            claimed.extend(w.keys.iter().map(String::as_str));
        }
        self.room(place.class, jobs, want, &ahead) && self.free(&want.keys, &claimed)
    }

    /// Whether a run of `class` that `want` has a job among `jobs` free,
    /// its owner holding fewer than `want` may, with those `ahead` of it.
    fn room(
        &self,
        class: Class,
        jobs: (usize, usize),
        want: &Want,
        ahead: &HashMap<&str, usize>,
    ) -> bool {
        let owner = want.whose.owner.as_str();
        let holds = self.owners.get(owner).copied().unwrap_or_default();
        room(class, jobs) && holds + ahead.get(owner).copied().unwrap_or_default() < want.whose.most
    }

    fn free(&self, keys: &[String], claimed: &HashSet<&str>) -> bool {
        keys.iter()
            .all(|k| !self.held.contains(k) && !claimed.contains(k.as_str()))
    }

    /// The place in line of `owner`'s run ranked `rank`: the turn after
    /// its last waiting in the class started before it, or the last run's
    /// to start, if later. Of runs started together, the one first in
    /// line is then still the one started first.
    fn place(&self, rank: Rank, owner: &str) -> Place {
        let last = self.last.get(&rank.class).copied().unwrap_or_default();
        let mine = self
            .waiting
            .iter()
            .filter(|(p, w)| p.class == rank.class && p.seq < rank.seq && w.whose.owner == owner);
        let after = mine.map(|(p, _)| p.turn + 1).max().unwrap_or_default();
        Place {
            class: rank.class,
            turn: after.max(last),
            seq: rank.seq,
        }
    }
}

/// Whether a run of `class` has a job among `(jobs, others)` free.
fn room(class: Class, (jobs, others): (usize, usize)) -> bool {
    jobs > 0 && (class == Class::Main || others > 0)
}

/// `(jobs, others)` once a run of `class` has taken one.
fn taken(class: Class, (jobs, others): (usize, usize)) -> (usize, usize) {
    let other = usize::from(class != Class::Main);
    (jobs - 1, others - other)
}

impl Turns {
    /// `jobs` at once, `others` of them at most for runs other than
    /// `Main`.
    pub fn new(jobs: usize, others: usize) -> Arc<Turns> {
        Arc::new(Turns {
            board: Mutex::new(Board {
                jobs,
                others,
                held: HashSet::new(),
                owners: HashMap::new(),
                waiting: BTreeMap::new(),
                last: HashMap::new(),
            }),
            changed: Notify::new(),
        })
    }

    /// A job and `keys`, once the run ranked `rank`, `whose`, may have
    /// them. Dropped while it waits, it leaves the line.
    pub async fn take(self: Arc<Self>, rank: Rank, whose: Whose, mut keys: Vec<String>) -> Turn {
        keys.sort_unstable();
        keys.dedup();
        let place = {
            let mut board = self.board.lock().unwrap();
            let place = board.place(rank, &whose.owner);
            let want = Want { whose, keys };
            board.waiting.insert(place, want);
            place
        };
        let _line = Line {
            turns: &self,
            place,
        };
        loop {
            // Made before looking, so a change after the look still wakes it.
            let changed = self.changed.notified();
            if let Some(want) = self.start(&place) {
                return Turn {
                    turns: self.clone(),
                    class: place.class,
                    owner: want.whose.owner,
                    keys: want.keys,
                };
            }
            changed.await;
        }
    }

    /// Takes what the run at `place` waits for, if it may start, and
    /// gives back what that was.
    fn start(&self, place: &Place) -> Option<Want> {
        let mut board = self.board.lock().unwrap();
        let want = &board.waiting[place];
        if !board.starts(place, want) {
            return None;
        }
        let want = board.waiting.remove(place)?;
        (board.jobs, board.others) = taken(place.class, (board.jobs, board.others));
        board.held.extend(want.keys.iter().cloned());
        *board.owners.entry(want.whose.owner.clone()).or_default() += 1;
        let last = board.last.entry(place.class).or_default();
        *last = (*last).max(place.turn);
        Some(want)
    }
}

/// A run's place while it waits.
struct Line<'a> {
    turns: &'a Turns,
    place: Place,
}

impl Drop for Line<'_> {
    fn drop(&mut self) {
        let mut board = self.turns.board.lock().unwrap();
        if board.waiting.remove(&self.place).is_some() {
            drop(board);
            self.turns.changed.notify_waiters();
        }
    }
}

/// A job and keys held, let go when dropped.
pub struct Turn {
    turns: Arc<Turns>,
    class: Class,
    owner: String,
    keys: Vec<String>,
}

impl Drop for Turn {
    fn drop(&mut self) {
        let mut board = self.turns.board.lock().unwrap();
        board.jobs += 1;
        board.others += usize::from(self.class != Class::Main);
        for key in &self.keys {
            board.held.remove(key);
        }
        if let Some(holds) = board.owners.get_mut(&self.owner) {
            *holds -= 1;
        }
        drop(board);
        self.turns.changed.notify_waiters();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use futures::FutureExt;
    use std::pin::Pin;

    type Waiting = Pin<Box<dyn Future<Output = Turn>>>;

    fn rank(class: Class, seq: u64) -> Rank {
        Rank { class, seq }
    }

    fn wait(turns: &Arc<Turns>, rank: Rank, keys: &[&str]) -> Waiting {
        owned(turns, "o", rank, keys)
    }

    /// `owner`'s run waiting, whose owner may hold every job.
    fn owned(turns: &Arc<Turns>, owner: &str, rank: Rank, keys: &[&str]) -> Waiting {
        capped(turns, owner, usize::MAX, rank, keys)
    }

    /// `owner`'s run waiting, which goes while its owner holds fewer than
    /// `most` jobs.
    fn capped(turns: &Arc<Turns>, owner: &str, most: usize, rank: Rank, keys: &[&str]) -> Waiting {
        let keys = keys.iter().map(|k| k.to_string()).collect();
        let owner = owner.to_string();
        Box::pin(turns.clone().take(rank, Whose { owner, most }, keys))
    }

    /// The turn, if `waiting` has it now.
    fn ready(waiting: &mut Waiting) -> Option<Turn> {
        waiting.as_mut().now_or_never()
    }

    #[test]
    fn main_first_then_oldest() {
        let turns = Turns::new(1, 1);
        let going = ready(&mut wait(&turns, rank(Class::Other, 1), &[])).unwrap();
        let mut pull = wait(&turns, rank(Class::Other, 2), &[]);
        let mut later = wait(&turns, rank(Class::Other, 3), &[]);
        let mut main = wait(&turns, rank(Class::Main, 4), &[]);
        let mut asked = wait(&turns, rank(Class::Asked, 0), &[]);
        for w in [&mut pull, &mut later, &mut main, &mut asked] {
            assert!(ready(w).is_none());
        }
        // The run going is not stopped; once it ends, the push to main
        // goes before the older pull.
        drop(going);
        assert!(ready(&mut pull).is_none());
        let main = ready(&mut main).unwrap();
        assert!(ready(&mut later).is_none());
        drop(main);
        assert!(ready(&mut later).is_none());
        let pull = ready(&mut pull).unwrap();
        drop(pull);
        let later = ready(&mut later).unwrap();
        // One asked for goes after the rest, however old.
        assert!(ready(&mut asked).is_none());
        drop(later);
        assert!(ready(&mut asked).is_some());
    }

    #[test]
    fn later_after_the_rest() {
        let turns = Turns::new(1, 1);
        let going = ready(&mut wait(&turns, rank(Class::Other, 1), &[])).unwrap();
        let mut unqueued = wait(&turns, rank(Class::Later, 2), &[]);
        let mut pull = wait(&turns, rank(Class::Other, 3), &[]);
        for w in [&mut unqueued, &mut pull] {
            assert!(ready(w).is_none());
        }
        drop(going);
        assert!(ready(&mut unqueued).is_none());
        drop(ready(&mut pull).unwrap());
        assert!(ready(&mut unqueued).is_some());
    }

    #[test]
    fn keys_one_at_a_time() {
        let turns = Turns::new(4, 4);
        // Named twice, a key is taken once.
        let a = ready(&mut wait(&turns, rank(Class::Other, 1), &["v", "v"])).unwrap();
        let mut b = wait(&turns, rank(Class::Other, 2), &["v"]);
        assert!(ready(&mut b).is_none());
        // One with a key of its own goes beside it.
        let side = ready(&mut wait(&turns, rank(Class::Other, 3), &["w"]));
        assert!(side.is_some());
        drop(a);
        assert!(ready(&mut b).is_some());
    }

    #[test]
    fn claims_and_leftovers() {
        let turns = Turns::new(2, 2);
        let going = ready(&mut wait(&turns, rank(Class::Other, 1), &["v"])).unwrap();
        // Main waits for v, in use; w, which it needs too, is kept for it.
        let mut main = wait(&turns, rank(Class::Main, 2), &["v", "w"]);
        assert!(ready(&mut main).is_none());
        let mut wants_w = wait(&turns, rank(Class::Other, 3), &["w"]);
        assert!(ready(&mut wants_w).is_none());
        // One needing neither has the job left over.
        let mut free = wait(&turns, rank(Class::Other, 4), &["x"]);
        let free = ready(&mut free).unwrap();
        drop(free);
        drop(going);
        let main = ready(&mut main).unwrap();
        assert!(ready(&mut wants_w).is_none());
        drop(main);
        assert!(ready(&mut wants_w).is_some());
    }

    #[test]
    fn others_take_only_their_share() {
        let turns = Turns::new(3, 1);
        let pull = ready(&mut wait(&turns, rank(Class::Other, 1), &[])).unwrap();
        // A second pull waits, though jobs are free, and main goes past it.
        let mut second = wait(&turns, rank(Class::Other, 2), &[]);
        assert!(ready(&mut second).is_none());
        let main = ready(&mut wait(&turns, rank(Class::Main, 3), &[]));
        assert!(main.is_some());
        drop(pull);
        assert!(ready(&mut second).is_some());
    }

    #[test]
    fn leaving_the_line() {
        let turns = Turns::new(1, 1);
        let going = ready(&mut wait(&turns, rank(Class::Other, 1), &[])).unwrap();
        let mut main = wait(&turns, rank(Class::Main, 2), &["v"]);
        let mut pull = wait(&turns, rank(Class::Other, 3), &["v"]);
        assert!(ready(&mut main).is_none());
        assert!(ready(&mut pull).is_none());
        drop(going);
        // Cancelled while it waits, the one before leaves the line.
        drop(main);
        assert!(ready(&mut pull).is_some());
    }

    #[test]
    fn started_first_goes_first() {
        let turns = Turns::new(1, 1);
        let going = ready(&mut wait(&turns, rank(Class::Other, 1), &[])).unwrap();
        // Started in this order, but in line the other way round.
        let mut second = wait(&turns, rank(Class::Main, 3), &[]);
        assert!(ready(&mut second).is_none());
        let mut first = wait(&turns, rank(Class::Main, 2), &[]);
        assert!(ready(&mut first).is_none());
        drop(going);
        assert!(ready(&mut second).is_none());
        drop(ready(&mut first).unwrap());
        assert!(ready(&mut second).is_some());
    }

    #[test]
    fn owners_take_turns() {
        let turns = Turns::new(1, 1);
        let going = ready(&mut owned(&turns, "a", rank(Class::Other, 1), &[])).unwrap();
        // A burst of a's, then one of b's: b's goes after a's first
        // waiting, before the rest.
        let mut a2 = owned(&turns, "a", rank(Class::Other, 2), &[]);
        let mut a3 = owned(&turns, "a", rank(Class::Other, 3), &[]);
        let mut b4 = owned(&turns, "b", rank(Class::Other, 4), &[]);
        for w in [&mut a2, &mut a3, &mut b4] {
            assert!(ready(w).is_none());
        }
        drop(going);
        assert!(ready(&mut a3).is_none());
        assert!(ready(&mut b4).is_none());
        drop(ready(&mut a2).unwrap());
        assert!(ready(&mut a3).is_none());
        drop(ready(&mut b4).unwrap());
        assert!(ready(&mut a3).is_some());
    }

    #[test]
    fn an_owner_holds_only_its_share() {
        let turns = Turns::new(3, 3);
        let going = ready(&mut owned(&turns, "b", rank(Class::Other, 1), &[])).unwrap();
        let busy = ready(&mut owned(&turns, "c", rank(Class::Other, 2), &[])).unwrap();
        let more = ready(&mut owned(&turns, "c", rank(Class::Other, 3), &[])).unwrap();
        let mut a4 = capped(&turns, "a", 1, rank(Class::Other, 4), &[]);
        let mut a5 = capped(&turns, "a", 1, rank(Class::Other, 5), &[]);
        assert!(ready(&mut a4).is_none());
        assert!(ready(&mut a5).is_none());
        drop((going, busy, more));
        // Three jobs free, but a's first waiting takes a's one: its second
        // waits, and another owner's goes past it.
        assert!(ready(&mut a5).is_none());
        let a4 = ready(&mut a4).unwrap();
        assert!(ready(&mut a5).is_none());
        let b6 = ready(&mut owned(&turns, "b", rank(Class::Other, 6), &[]));
        assert!(b6.is_some());
        // What a run may hold is its own: one of a's that may hold two goes.
        let a7 = ready(&mut capped(&turns, "a", 2, rank(Class::Other, 7), &[]));
        assert!(a7.is_some());
        drop((a4, a7));
        assert!(ready(&mut a5).is_some());
    }
}