//! Landing a repository's merge queues (`knotless::queue`), with
//! `--queue`, as its owner or a collaborator.
//!
//! Each branch's queue goes a batch at a time: up to `--batch` queued
//! pulls in order, whatever their own checks say, each squashed onto the
//! branch's tip as a commit of its own on the one before. The workflows
//! that run on a pull into the branch run once, at the last of them, filed
//! as checks keyed by the queue that name the batch. Once they all pass the
//! branch moves there, from where the batch was built and only from there,
//! and each pull is marked merged.
//!
//! A pull goes into a batch only once its pre-flights, the workflows that
//! say `queue: first`, have passed at its head, as any runner the
//! repository names filed them: it waits while one runs, holding up none
//! behind it, and is ejected without a batch once one fails on the
//! branch's tip. One nobody ran there is run here, as a pull's run.
//!
//! A pull that does not apply on what is before it is ejected, naming the
//! paths, and the batch goes on without it. A batch that fails is halved,
//! its first half run alone, until the pull that breaks the branch is found
//! and ejected: a half that passes lands, and what is left of a batch that
//! failed is then the same tree on the tip it moved to, so it fails too,
//! and a lone pull left is ejected without running again. The branch
//! moving any other way, or a pull in a batch moving or leaving, drops the
//! batches going, to be built again.
//!
//! While a batch runs, the next is built on its tip and run, as if it
//! passed (`--depth` 2, the default; 1 runs one at a time). It runs ahead
//! as a pull does: after the runs of pushes to the default branch and of
//! the batch it is on, with a job left over and the CPU share a pull's run
//! has, so it never holds up a deploy. The batch it is on failing, or
//! changing, drops it, cancelling its runs; it lands only on the commit
//! that batch landed as, and nothing lands but on the very commit it was
//! built on.
//!
//! The queue records are read from the repos of those who may write them,
//! the owner's and each collaborator's, not only as the appview has
//! indexed them, and looked at every `--notice` besides as Jetstream
//! brings them: a pull queued is taken in that long at most.

use crate::telemetry;
use crate::{
    Asks, DROPPED, Pull, Runner, Subject, Watch, cancel, changes, short, vouched, workflow,
};
use anyhow::{Context, Result, bail};
use knotless::approvals;
use knotless::checks::{self, Filter};
use knotless::forge::{self, Item, Target};
use knotless::git::Git;
use knotless::keys;
use knotless::mirror::Mirror;
use knotless::queue::{self, Stack};
use knotless::records::queue::{CONFLICTS, HALVED};
use knotless::records::{Batched, FailedCheck, RepoRef, nsid, status};
use knotless::remote::Snapshot;
use knotless::session::{self, Session};
use knotless::xrpc::XrpcError;
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::sync::{Arc, Mutex};

/// Why landing was refused, when the gateway refused it: a runner that
/// lands as a runner the repository names, through the gateway, is refused
/// until the owner delegates writes to it.
fn refused(err: &anyhow::Error) -> Option<String> {
    let refused = XrpcError::of(err).filter(|e| e.status == 403)?;
    Some(format!(
        "the gateway would not land its batch: {}. A runner lands the merge queue through the gateway the owner delegates writes to: under Delegated writes in their settings",
        refused.message
    ))
}

/// A batch a run is the tip of: the branch, in full, its tip the batch is
/// on, the pulls in it, and how each workflow's run of it ended.
#[derive(Clone)]
pub(crate) struct Batch {
    pub base: String,
    pub base_commit: String,
    pub pulls: Vec<Batched>,
    /// Whether everyone whose code is in it may push here: then it shares
    /// the pushes' caches.
    pub trusted: bool,
    /// Whether it was built on a batch still being checked: its runs go
    /// as a pull's do.
    pub ahead: bool,
    /// The accounts whose shared steps its lock adds that the repository
    /// does not vouch for, which make it untrusted.
    pub strangers: Vec<String>,
    runs: Arc<Mutex<BTreeMap<String, Option<String>>>>,
}

impl Batch {
    pub fn new(base: &str, base_commit: &str, pulls: Vec<Batched>, trusted: bool) -> Batch {
        Batch {
            base: base.to_string(),
            base_commit: base_commit.to_string(),
            pulls,
            trusted,
            ahead: false,
            strangers: Vec::new(),
            runs: Arc::default(),
        }
    }

    /// Notes a run of `workflow` started.
    pub fn started(&self, workflow: &str) {
        self.runs.lock().unwrap().insert(workflow.to_string(), None);
    }

    /// Notes how the run of `workflow` ended.
    pub fn ended(&self, workflow: &str, status: &str) {
        let ended = Some(status.to_string());
        self.runs
            .lock()
            .unwrap()
            .insert(workflow.to_string(), ended);
    }

    /// How its runs ended, once they all have.
    pub fn outcome(&self) -> Option<BTreeMap<String, String>> {
        let runs = self.runs.lock().unwrap();
        runs.iter()
            .map(|(w, s)| Some((w.clone(), s.clone()?)))
            .collect()
    }
}

/// A batch going: how it was built, and its runs.
struct Flight {
    stack: Stack,
    batch: Batch,
}

/// A branch's merge queue as the runner lands it: the batches going, the
/// lowest on the branch's tip and each after on the one before; once a
/// batch has failed, its pulls, one of which breaks the branch, and how it
/// failed; and the pulls landed or ejected, by the key of the record that
/// queued them, until the queue as read says so too; and the pulls whose
/// pre-flights it ran itself, each at what (see [`at`](crate::at)).
#[derive(Default)]
pub(crate) struct Line {
    flights: Vec<Flight>,
    suspects: Vec<String>,
    failed: (String, Vec<FailedCheck>),
    done: HashMap<String, String>,
    flown: HashMap<String, String>,
}

impl Line {
    /// The pulls the batches in flight hold, by URI.
    fn flying(&self) -> impl Iterator<Item = &str> {
        (self.flights.iter()).flat_map(|f| f.stack.stacked.iter().map(|s| s.pull.uri.as_str()))
    }
}

/// How a pull's pre-flights stand at its head.
#[derive(Default)]
struct Preflight {
    /// The newest run of each that did not pass on the branch's tip.
    failed: Vec<checks::Check>,
    running: bool,
    /// Those with no run there, or only one cancelled or one that failed
    /// on another tip.
    missing: BTreeSet<String>,
}

/// A repository whose merge queues a runner lands, as last read, and the
/// queue and state records about it, each by URI and what it says, as
/// last looked at.
#[derive(Default)]
pub(crate) struct Noticing {
    target: Option<Arc<Target>>,
    seen: Option<BTreeSet<String>>,
}

/// The key of the record that queued `item`.
fn queued_at(item: &Item) -> &str {
    item.queue.as_ref().map_or("", |q| q.at.as_str())
}

/// How the pre-flights `first` of a pull stand at its head, by its runs
/// there, `runs`, newest first, on the branch at `tip`. Another runner's
/// run only queued there counts as none: it may wait long for a job, so
/// this runner, `me`, runs it too, and whichever starts first runs it.
fn preflight(first: &BTreeSet<String>, runs: &[checks::Check], tip: &str, me: &str) -> Preflight {
    let mut stands = Preflight::default();
    let counts = |c: &&checks::Check| c.runner == me || c.record.status != status::QUEUED;
    for workflow in first {
        let newest = (runs.iter().filter(counts)).find(|c| c.record.workflow == *workflow);
        match newest.map(|c| (c, c.record.status.as_str())) {
            Some((_, status::SUCCESS)) => {}
            Some((_, status::RUNNING | status::QUEUED)) => stands.running = true,
            Some((c, status::FAILURE | status::ERROR))
                if c.record.base_commit.as_deref() == Some(tip) =>
            {
                stands.failed.push(c.clone());
            }
            _ => {
                stands.missing.insert(workflow.clone());
            }
        }
    }
    stands
}

/// Why a pull that conflicts in `files`, built on `into` at `tip`, is
/// ejected: naming the pulls ahead of it that change them, `touching`,
/// and how to take theirs in, or else the branch itself.
fn conflicted(into: &str, tip: &str, files: &[String], touching: &[&queue::Stacked]) -> String {
    let files = files.join(", ");
    if touching.is_empty() {
        return format!(
            "it does not apply on {into} at {}: {CONFLICTS} {files}. Merge {into} into it, or rebase it onto {into}, and queue it again",
            short(tip)
        );
    }
    let named: Vec<String> = touching
        .iter()
        .map(|s| {
            let branch = s.pull.head.as_ref().map_or("", |h| h.branch.as_str());
            format!("{} ({branch})", s.pull.uri)
        })
        .collect();
    format!(
        "{CONFLICTS} {files} with {}, queued ahead of it on {into} at {}. Merge that branch into it, or wait for it to land and merge {into} into it, and queue it again",
        named.join(" and "),
        short(tip)
    )
}

impl Runner {
    /// Lands what the merge queues of `watch` can, ejects what they must,
    /// and starts their next batches. A runner that may not push to the
    /// repository leaves its queues be.
    pub(crate) async fn queues(
        self: &Arc<Self>,
        watch: &Watch,
        repo: &RepoRef,
        snap: &Snapshot,
    ) -> Result<()> {
        let target = Arc::new(Target::open(&self.ctx, &watch.remote.url).await?);
        if !self.lands(&target.record, watch) {
            return Ok(());
        }
        {
            let mut noticing = self.noticing.lock().unwrap();
            noticing.entry(watch.id()).or_default().target = Some(target.clone());
        }
        let me = session::me(&self.ctx).await?;
        let entries = target.queue_entries(&self.ctx).await?;
        let items = forge::assemble(&target.record, watch.remote.did(), &entries);
        let lines = queue::by_branch(&items);
        let prefix = format!("{}\n", watch.id());
        let mut branches: BTreeSet<String> = lines.keys().cloned().collect();
        let known: Vec<String> = self.lines.lock().unwrap().keys().cloned().collect();
        let mine = known.iter().filter_map(|k| k.strip_prefix(&prefix));
        branches.extend(mine.map(str::to_string));
        if branches.is_empty() {
            return Ok(());
        }
        watch.at("syncing the mirror");
        let mirror = Mirror::open(&self.ctx, watch.remote.did(), &watch.key).await?;
        mirror.sync(&watch.remote, snap).await?;
        let none = Vec::new();
        for branch in branches {
            let on = Queue {
                watch,
                repo,
                snap,
                target: &target,
                git: &mirror.git,
                me: &me,
                branch: &branch,
            };
            self.line(&on, lines.get(&branch).unwrap_or(&none)).await?;
        }
        Ok(())
    }

    /// Lands, as the runner stops, the batches of `watch` whose runs have
    /// all passed, and does nothing else: no batch is started or ejected,
    /// and what failed is checked again on the next start.
    ///
    /// Boxed: it is awaited in the task of the run that ends, and starts
    /// runs itself, a cycle in which the compiler cannot see it is `Send`.
    pub(crate) fn land_passed<'a>(
        self: &'a Arc<Self>,
        watch: &'a Watch,
    ) -> futures::future::BoxFuture<'a, Result<()>> {
        Box::pin(async move {
            let _one = self.landing.lock().await;
            watch.at("reading the repository");
            let snap = watch.remote.snapshot(&self.ctx).await?;
            let repo = snap
                .reference(watch.remote.did())
                .context("a snapshot with a key has a repository")?;
            self.queues(watch, &repo, &snap).await
        })
    }

    /// Goes on with the queue of `on.branch`, whose pulls queued are
    /// `queued`, in order.
    async fn line(self: &Arc<Self>, on: &Queue<'_>, queued: &[Item]) -> Result<()> {
        let id = format!("{}\n{}", on.watch.id(), on.branch);
        let mut line = self.lines.lock().unwrap().remove(&id).unwrap_or_default();
        let went = self.go_on(on, queued, &mut line).await;
        let idle = line.flights.is_empty()
            && line.suspects.is_empty()
            && line.done.is_empty()
            && line.flown.is_empty();
        if !idle {
            self.lines.lock().unwrap().insert(id, line);
        }
        went
    }

    async fn go_on(self: &Arc<Self>, on: &Queue<'_>, all: &[Item], line: &mut Line) -> Result<()> {
        // What this runner landed or ejected, until the queue as read has
        // caught up.
        let uris: HashSet<&str> = all.iter().map(|i| i.uri.as_str()).collect();
        line.done.retain(|uri, _| uris.contains(uri.as_str()));
        line.flown.retain(|uri, _| uris.contains(uri.as_str()));
        let done = |i: &&Item| line.done.get(&i.uri).map(String::as_str) == Some(queued_at(i));
        let queued: Vec<Item> = all.iter().filter(|i| !done(i)).cloned().collect();
        let refname = format!("refs/heads/{}", on.branch);
        let found = on.snap.find_ref(&refname);
        let Some(mut tip) = found.and_then(|r| r.value.target.clone()) else {
            self.drop_from(on.watch, line, 0);
            return Ok(());
        };
        let generation = found.map_or(0, |r| r.value.generation);
        // The branch moved some other way: what is going is of a tip it is
        // no longer at, and what failed there may not fail now.
        if line.flights.first().is_some_and(|f| f.stack.base != tip) {
            self.drop_from(on.watch, line, 0);
            line.suspects.clear();
        }
        // A pull in a batch, or among those suspected, left the queue.
        let still: HashSet<&str> = queued.iter().map(|i| i.uri.as_str()).collect();
        let stays = |uri: &String| still.contains(uri.as_str());
        let left = |f: &Flight| !f.stack.stacked.iter().all(|s| stays(&s.pull.uri));
        if let Some(at) = line.flights.iter().position(left) {
            self.drop_from(on.watch, line, at);
        }
        if !line.suspects.iter().all(stays) {
            line.suspects.clear();
        }
        while let Some(ended) = line.flights.first().and_then(|f| f.batch.outcome()) {
            let flight = line.flights.remove(0);
            let failed: Vec<String> = ended
                .iter()
                .filter(|(_, s)| *s != status::SUCCESS)
                .map(|(w, s)| format!("{w} ({s})"))
                .collect();
            let checks = self.failed_checks(on, &flight, &ended);
            if failed.is_empty() {
                on.watch.at("landing a batch");
                if let Err(err) = self.land(on, &flight).await {
                    let refused = refused(&err);
                    let (label, branch) = (&on.watch.label, on.branch);
                    telemetry::failed(label, &err.context(format!("landing on {branch}")));
                    self.drop_from(on.watch, line, 0);
                    // Refused, it would be refused again: its pulls go,
                    // rather than be checked and refused for ever.
                    if let Some(why) = &refused {
                        for s in &flight.stack.stacked {
                            self.eject(on, line, &s.pull, Some(&s.head), why, &[])
                                .await?;
                        }
                        line.suspects.clear();
                    }
                    break;
                }
                tip = flight.stack.tip().to_string();
                let landed = flight.stack.stacked.len();
                telemetry::batch(
                    "batch landed",
                    &on.watch.label,
                    on.branch,
                    &tip,
                    landed,
                    None,
                );
                for s in &flight.stack.stacked {
                    let at = queued_at(&s.pull).to_string();
                    line.done.insert(s.pull.uri.clone(), at);
                    line.suspects.retain(|u| *u != s.pull.uri);
                }
                // What is left of a batch that failed is the same tree on
                // the tip it moved to, and fails as it did: one alone is
                // what broke it.
                if let [lone] = &line.suspects[..] {
                    let item = queued.iter().find(|i| &i.uri == lone);
                    let item = item.expect("a suspect is queued").clone();
                    let (why, checks) = line.failed.clone();
                    self.eject(on, line, &item, None, &why, &checks).await?;
                    line.suspects.clear();
                }
                continue;
            }
            self.drop_from(on.watch, line, 0);
            let why = format!(
                "its batch failed its checks at {}, on {} at {}: {}",
                short(flight.stack.tip()),
                on.branch,
                short(&flight.stack.base),
                failed.join(", "),
            );
            match &flight.stack.stacked[..] {
                [one] => {
                    self.eject(on, line, &one.pull, Some(&one.head), &why, &checks)
                        .await?;
                    line.suspects.clear();
                }
                all => {
                    line.suspects = all.iter().map(|s| s.pull.uri.clone()).collect();
                    line.failed = (format!("{why}. {HALVED}"), checks);
                }
            }
        }
        // Stopping, it lands what passed and starts nothing.
        if *self.stopping.borrow() {
            return Ok(());
        }
        // Of the pulls waiting, only those whose pre-flights have passed go
        // into a batch; the rest wait, or are ejected, holding up none
        // behind them.
        let going: HashSet<&str> = (line.flying())
            .chain(line.suspects.iter().map(String::as_str))
            .collect();
        let outside: Vec<Item> = (queued.iter())
            .filter(|i| !line.done.contains_key(&i.uri) && !going.contains(i.uri.as_str()))
            .cloned()
            .collect();
        on.watch.at("checking pre-flights");
        let ready = self.preflown(on, line, &tip, outside).await?;
        while line.flights.len() < self.opts.depth {
            // Halving is not run ahead of.
            if !(line.suspects.is_empty() || line.flights.is_empty()) {
                break;
            }
            let ahead = !line.flights.is_empty();
            let on_tip = line
                .flights
                .last()
                .map_or(tip.clone(), |f| f.stack.tip().to_string());
            let taken: HashSet<&str> = line.flying().collect();
            let half = &line.suspects[..line.suspects.len().div_ceil(2)];
            let waiting = queued.iter().filter(|i| !line.done.contains_key(&i.uri));
            let next: Vec<Item> = match half {
                [] => waiting
                    .filter(|i| !taken.contains(i.uri.as_str()) && ready.contains(&i.uri))
                    .take(self.opts.batch)
                    .cloned()
                    .collect(),
                half => waiting.filter(|i| half.contains(&i.uri)).cloned().collect(),
            };
            if next.is_empty() {
                break;
            }
            let mut pulls = Vec::new();
            for item in next {
                let head = item.head.as_ref().expect("a pull has a head");
                match forge::head_of(&self.ctx, on.git, head).await? {
                    Ok(sha) => pulls.push((item, sha)),
                    Err(why) => self.eject(on, line, &item, None, &why, &[]).await?,
                }
            }
            on.watch.at("stacking a batch");
            let stack = queue::stack(on.git, &on_tip, &pulls, on.me).await?;
            // Ahead of a batch that may yet fail, a conflict may not stand:
            // the pull waits to be built on the branch itself.
            for (item, head, files) in stack.conflicts.iter().filter(|_| !ahead) {
                let touching = stack.touching(on.git, files).await?;
                let why = conflicted(on.branch, &on_tip, files, &touching);
                self.eject(on, line, item, Some(head), &why, &[]).await?;
            }
            if stack.stacked.is_empty() {
                if ahead {
                    break;
                }
                continue;
            }
            on.watch.at("launching a batch");
            let flight = self.launch(on, stack, generation, ahead).await?;
            line.flights.push(flight);
        }
        Ok(())
    }

    /// Whether the queue and state records about `watch` that those who
    /// may write them have, the newest of each, are others than the last
    /// look found: the first look, too. Nothing, until a pass has found
    /// this runner lands its queues.
    pub async fn noticed(&self, watch: &Watch) -> Result<bool> {
        let noticing = self
            .noticing
            .lock()
            .unwrap()
            .get(&watch.id())
            .map(|n| n.target.clone());
        let Some(target) = noticing.flatten() else {
            return Ok(false);
        };
        let fresh = target.fresh_entries(&self.ctx).await;
        let found = fresh.iter().map(|e| format!("{} {}", e.uri, e.value));
        let found: BTreeSet<String> = found.collect();
        let mut noticing = self.noticing.lock().unwrap();
        let seen = &mut noticing.entry(watch.id()).or_default().seen;
        Ok(seen.replace(found.clone()) != Some(found))
    }

    /// The pulls among `waiting` that may go into a batch on the branch at
    /// `tip`: each whose pre-flights, the workflows at `tip` that run
    /// first in its queue, have each passed at its head, on any tip, as any
    /// runner the repository names, or this one, filed it. One whose
    /// newest run of one failed on `tip` is ejected; one with a pre-flight
    /// that has none, or one cancelled or failed on another tip, has it
    /// run here, as a pull's run, and waits. Once this runner has run them
    /// at its head on `tip`, it waits no longer: it is a workflow the pull
    /// takes away, say, and its batch will say what it does. Into a branch
    /// whose rule asks for reviews, one not reviewed as it asks is ejected
    /// first, as a merge would refuse it.
    async fn preflown(
        self: &Arc<Self>,
        on: &Queue<'_>,
        line: &mut Line,
        tip: &str,
        waiting: Vec<Item>,
    ) -> Result<HashSet<String>> {
        let waiting = self.reviewed(on, line, tip, waiting).await?;
        let refname = format!("refs/heads/{}", on.branch);
        let loaded = workflow::load(on.git, tip).await?;
        let first: BTreeSet<String> = (loaded.into_iter())
            .filter_map(|l| l.workflow.ok())
            .filter(|w| w.first_in_queue(&refname))
            .map(|w| w.name)
            .collect();
        if first.is_empty() {
            return Ok(waiting.into_iter().map(|i| i.uri).collect());
        }
        let mut ready = HashSet::new();
        let (record, me) = (&on.target.record, &on.me.did);
        let mut runners: Vec<String> = record.runners.iter().map(|r| r.did.clone()).collect();
        runners.extend((!runners.contains(me)).then_some(me.clone()));
        for item in waiting {
            let head = item.head.as_ref().expect("a pull has a head");
            // One whose branch is gone is ejected as a batch is built.
            let found = forge::head_at(&self.ctx, on.git, head).await?;
            let Ok((sha, generation)) = found else {
                ready.insert(item.uri);
                continue;
            };
            let filter = Filter {
                pull: Some(&item.uri),
                commit: Some(&sha),
                ..Filter::default()
            };
            let (remote, key) = (&on.target.remote, &record.key);
            let runs = checks::by(&self.ctx, remote, key, &runners, filter, false).await?;
            let stands = preflight(&first, &runs, tip, me);
            if !stands.failed.is_empty() {
                let failed: Vec<String> = (stands.failed.iter())
                    .map(|c| format!("{} ({})", c.record.workflow, c.record.status))
                    .collect();
                let why = format!(
                    "its pre-flight failed at its head {}, on {} at {}: {}",
                    short(&sha),
                    on.branch,
                    short(tip),
                    failed.join(", "),
                );
                let checks: Vec<FailedCheck> = (stands.failed.iter())
                    .map(|c| FailedCheck {
                        workflow: c.record.workflow.clone(),
                        status: Some(c.record.status.clone()),
                        check: Some(c.uri.clone()),
                    })
                    .collect();
                self.eject(on, line, &item, Some(&sha), &why, &checks)
                    .await?;
                continue;
            }
            if stands.running {
                continue;
            }
            let at = crate::at(&sha, Some(tip));
            if !stands.missing.is_empty() && line.flown.get(&item.uri) != Some(&at) {
                let only = stands.missing;
                let flew = self.fly(on, &item, (&sha, generation), tip, only).await?;
                line.flown.insert(item.uri.clone(), at);
                // One that does not merge goes on, to be ejected as a batch
                // is built, saying where it conflicts.
                if flew {
                    continue;
                }
            }
            ready.insert(item.uri);
        }
        Ok(ready)
    }

    /// Of `waiting`, those reviewed as their branch's rule asks, on the
    /// branch at `tip`; the rest are ejected, saying what they lack. One
    /// whose branch is gone goes on, to be ejected as a batch is built.
    async fn reviewed(
        &self,
        on: &Queue<'_>,
        line: &mut Line,
        tip: &str,
        waiting: Vec<Item>,
    ) -> Result<Vec<Item>> {
        let rule = on.target.record.rule(on.branch);
        if rule.is_none_or(|r| r.reviews.is_none()) {
            return Ok(waiting);
        }
        let mut reviewed = Vec::new();
        for item in waiting {
            let head = item.head.as_ref().expect("a pull has a head");
            let Ok(sha) = forge::head_of(&self.ctx, on.git, head).await? else {
                reviewed.push(item);
                continue;
            };
            let standing =
                approvals::of(&self.ctx, on.target.facts(), &item, on.git, tip, &sha).await?;
            let url = &on.watch.remote.url;
            let why = standing.and_then(|s| approvals::refusal(url, &item.uri, on.branch, &s));
            let eject = why
                .as_deref()
                .map(|why| self.eject(on, line, &item, Some(&sha), why, &[]));
            let ejected = futures::future::OptionFuture::from(eject).await;
            let ejected = ejected.transpose()?;
            reviewed.extend(ejected.is_none().then_some(item));
        }
        Ok(reviewed)
    }

    /// Runs the pre-flights `only` of `item`, at its `head` (and the
    /// branch's generation there) merged with the branch at `tip`, as a
    /// pull's run, unless the two conflict: whether it does.
    async fn fly(
        self: &Arc<Self>,
        on: &Queue<'_>,
        item: &Item,
        (head, generation): (&str, u64),
        tip: &str,
        only: BTreeSet<String>,
    ) -> Result<bool> {
        let merged = forge::merge_result(on.git, tip, head).await?;
        let Ok(merge) = &merged else {
            return Ok(false);
        };
        let changed = changes(on.git, tip, merge).await;
        let (record, owner) = (&on.target.record, on.watch.remote.did());
        let branch = item.head.as_ref().expect("a pull has a head");
        let trusted =
            vouched(record, owner, &item.author) && vouched(record, owner, branch.repo.did());
        let strangers = self.strangers(on.git, record, owner, tip, merge).await;
        let trusted = trusted && strangers.is_empty();
        // Where the pull ran, so a pass over the pulls finds it there, and
        // does not cancel it as moved.
        let id = on.watch.id();
        if let Some(pulls) = self.state.lock().unwrap().pulls.get_mut(&id) {
            pulls.insert(item.uri.clone(), crate::at(head, Some(tip)));
        }
        let subject = Subject {
            ref_name: format!("refs/heads/{}", branch.branch),
            commit: head.to_string(),
            generation,
            main: false,
            pull: Some(Pull {
                uri: item.uri.clone(),
                base: format!("refs/heads/{}", on.branch),
                base_commit: tip.to_string(),
                // Only a --pulls-only repository's pulls warm its caches,
                // and its queue is landed elsewhere.
                main_generation: None,
                merged,
                trusted,
                later: false,
                strangers,
                others: self.others(record),
            }),
            tree: None,
            changed,
            batch: None,
            pushed: true,
            asked: Asks::new(),
            only: Some(only),
        };
        self.run_at(on.watch, on.repo, on.git, subject).await?;
        Ok(true)
    }

    /// Starts the runs of a batch built as `stack`, `ahead` of one still
    /// being checked or not.
    async fn launch(
        self: &Arc<Self>,
        on: &Queue<'_>,
        stack: Stack,
        generation: u64,
        ahead: bool,
    ) -> Result<Flight> {
        let (record, owner) = (&on.target.record, on.watch.remote.did());
        let trusted = stack.stacked.iter().all(|s| {
            let head = s.pull.head.as_ref().map_or("", |h| h.repo.did());
            vouched(record, owner, &s.pull.author) && vouched(record, owner, head)
        });
        let strangers = self.strangers(on.git, record, owner, &stack.base, stack.tip());
        let strangers = strangers.await;
        let trusted = trusted && strangers.is_empty();
        let pulls = stack.stacked.iter().map(|s| Batched {
            pull: s.pull.uri.clone(),
            commit: s.commit.clone(),
            head: Some(s.head.clone()),
        });
        let refname = format!("refs/heads/{}", on.branch);
        let mut batch = Batch::new(&refname, &stack.base, pulls.collect(), trusted);
        batch.ahead = ahead;
        batch.strangers = strangers;
        let changed = changes(on.git, &stack.base, stack.tip()).await;
        let subject = Subject {
            ref_name: refname,
            commit: stack.tip().to_string(),
            generation,
            main: true,
            pull: None,
            tree: None,
            changed,
            batch: Some(batch.clone()),
            pushed: true,
            asked: Asks::new(),
            only: None,
        };
        self.run_at(on.watch, on.repo, on.git, subject).await?;
        let said = ["batch started", "batch started, ahead of another"][usize::from(ahead)];
        let (tip, pulls) = (stack.tip(), stack.stacked.len());
        telemetry::batch(said, &on.watch.label, on.branch, tip, pulls, None);
        Ok(Flight { stack, batch })
    }

    /// Lands `flight`, if each pull in it is still at the head it took.
    async fn land(&self, on: &Queue<'_>, flight: &Flight) -> Result<()> {
        for s in &flight.stack.stacked {
            let head = s.pull.head.as_ref().expect("a pull has a head");
            let now = forge::head_of(&self.ctx, on.git, head).await?;
            if now.as_ref() != Ok(&s.head) {
                bail!("{} moved on from {}", s.pull.uri, short(&s.head));
            }
        }
        let url = &on.watch.remote.url;
        queue::land(&self.ctx, url, on.git, on.branch, &flight.stack).await
    }

    /// The runs of `flight` that `ended` other than passing, each with its
    /// check, as this runner filed it.
    fn failed_checks(
        &self,
        on: &Queue<'_>,
        flight: &Flight,
        ended: &BTreeMap<String, String>,
    ) -> Vec<FailedCheck> {
        let (owner, key) = (on.watch.remote.did(), &on.watch.key);
        let name = keys::queue_name(&flight.batch.base);
        let (tip, base) = (flight.stack.tip(), Some(flight.stack.base.as_str()));
        let failed = ended.iter().filter(|(_, s)| *s != status::SUCCESS);
        failed
            .map(|(w, s)| {
                let rkey = keys::check_rkey(owner, key, &name, tip, base, w);
                FailedCheck {
                    workflow: w.clone(),
                    status: Some(s.clone()),
                    check: Some(
                        knotless::aturi::AtUri::public(&on.me.did, nsid::CHECK, &rkey)
                            .resolved(on.watch.remote.space_ref().as_ref())
                            .to_string(),
                    ),
                }
            })
            .collect()
    }

    /// Ejects `item` from its queue, saying `why`, at `head` if it has one.
    async fn eject(
        &self,
        on: &Queue<'_>,
        line: &mut Line,
        item: &Item,
        head: Option<&str>,
        why: &str,
        checks: &[FailedCheck],
    ) -> Result<()> {
        let ejected = queue::Ejected {
            reason: Some(why),
            head,
            checks,
        };
        on.watch.at("ejecting a pull");
        queue::eject(&self.ctx, on.target, &item.uri, ejected).await?;
        telemetry::ejected(&on.watch.label, &item.uri, why);
        let at = queued_at(item).to_string();
        line.done.insert(item.uri.clone(), at);
        Ok(())
    }

    /// Drops `line`'s batches from the `at`th up, cancelling their runs.
    fn drop_from(&self, watch: &Watch, line: &mut Line, at: usize) {
        let dropped: Vec<Flight> = line.flights.drain(at.min(line.flights.len())..).collect();
        let going = self.going.lock().unwrap();
        let repo = format!("{}\n", watch.id());
        for flight in dropped {
            let (tip, pulls) = (flight.stack.tip(), flight.stack.stacked.len());
            telemetry::batch(
                "batch dropped",
                &watch.label,
                &flight.batch.base,
                tip,
                pulls,
                None,
            );
            let at = format!("{} {}", flight.stack.tip(), flight.stack.base);
            let runs = going
                .iter()
                .filter(|(id, g)| id.starts_with(&repo) && g.at == at);
            for (_, g) in runs {
                cancel(&g.cancel, DROPPED);
            }
        }
    }
}

/// A branch's merge queue, and where it is landed from.
struct Queue<'a> {
    watch: &'a Watch,
    repo: &'a RepoRef,
    snap: &'a Snapshot,
    target: &'a Target,
    git: &'a Git,
    me: &'a Session,
    branch: &'a str,
}

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

    #[test]
    fn how_preflights_stand() {
        let by = |runner: &str, workflow: &str, status: &str, base: &str| checks::Check {
            uri: format!("at://{runner}/dev.knotless.check/{workflow}"),
            runner: runner.into(),
            record: serde_json::from_value(serde_json::json!({
                "repo": {"uri": "at://did:a/dev.knotless.repo/r", "key": "k"},
                "ref": "refs/heads/x", "commit": "h", "baseCommit": base,
                "workflow": workflow, "status": status, "startedAt": "2026-01-01T00:00:00Z",
            }))
            .unwrap(),
            log_text: None,
            space: None,
        };
        let run = |workflow: &str, status: &str, base: &str| by("did:ci", workflow, status, base);
        let first = |all: &[&str]| all.iter().map(|w| w.to_string()).collect::<BTreeSet<_>>();
        // Passed on any tip, it is ready.
        let stands = preflight(
            &first(&["a"]),
            &[run("a", "success", "old")],
            "tip",
            "did:ci",
        );
        assert!(!stands.running && stands.failed.is_empty() && stands.missing.is_empty());
        // The newest run counts: one running now.
        let runs = [run("a", "running", "tip"), run("a", "failure", "tip")];
        let stands = preflight(&first(&["a"]), &runs, "tip", "did:ci");
        assert!(stands.running && stands.failed.is_empty());
        // Failed on the tip, it is failed; on another, or cancelled, or
        // never run, it is to run again.
        let runs = [
            run("a", "failure", "tip"),
            run("b", "error", "old"),
            run("c", "cancelled", "tip"),
        ];
        let stands = preflight(&first(&["a", "b", "c", "d"]), &runs, "tip", "did:ci");
        let failed: Vec<&str> = stands
            .failed
            .iter()
            .map(|c| c.record.workflow.as_str())
            .collect();
        assert_eq!(failed, ["a"]);
        assert_eq!(stands.missing, first(&["b", "c", "d"]));
        // Its own run queued, it waits; another runner's, it runs it too,
        // unless an older run there is going.
        let runs = [run("a", "queued", "tip")];
        assert!(preflight(&first(&["a"]), &runs, "tip", "did:ci").running);
        let runs = [by("did:other", "a", "queued", "tip")];
        let stands = preflight(&first(&["a"]), &runs, "tip", "did:ci");
        assert!(!stands.running && stands.missing == first(&["a"]));
        let runs = [
            by("did:other", "a", "queued", "tip"),
            run("a", "running", "tip"),
        ];
        assert!(preflight(&first(&["a"]), &runs, "tip", "did:ci").running);
    }

    #[test]
    fn a_conflict_with_the_branch_itself_says_so() {
        let why = conflicted("main", &"a".repeat(40), &["f".into(), "g".into()], &[]);
        assert_eq!(
            why,
            "it does not apply on main at aaaaaaaaaaaa: it conflicts in f, g. Merge main into it, or rebase it onto main, and queue it again"
        );
    }
}