//! The reference sink scheduler: a sink's run of its plans (`extensions/timeliness.md` §7.5).
//!
//! From a sink's plans (§7) and its own clock — the instants from which it first holds each
//! object — [`run`] decides what the sink fetches, what it presents in each object's place and
//! when, and what it states, by the rule a sink is graded by (§5.3), the ladder included. A run is
//! a function of its inputs alone, so a verifier given a sink's inputs derives its decisions.
//!
//! # Rules enforced here rather than only documented
//!
//! - At an object's deadline the sink presents the primary it holds, else the first rung of the
//!   ladder it holds; holding neither, it has passed every rung and says so (`passed`), and no rung
//!   is presented in the primary's place from then on.
//! - A late primary is presented at its value at its arrival, where that is at least 1; at the
//!   end, a sink that presented nothing abandons its fetch and says so (`ended`).
//! - A later plan replaces the one in force wholly: a fetch it does not continue is abandoned, and
//!   an object only the earlier plan wanted is no longer run. What is settled stays settled.
//! - Each plan reads the groups, object receipts, envelopes, claims and records the sink held by its
//!   own instant, and counts the symbols it held then. A statement the sink holds later decides
//!   nothing until a plan reads it, and what the sink holds stays held whatever a later claim sets
//!   aside.

use std::collections::{BTreeMap, BTreeSet};

use serde_json::Value;

use super::{checked, planned, position_ok, wanted_formed, At, Choice, ClaimVerdict, GroupInput, PlanInputs, Refusal, Source, SymbolsHeld, Which, SYMBOLS_INVALID, WANTED_INVALID};
use crate::bearer_receipts::HeldObject;
use crate::claim_receipt::schedule::LAST_INSTANT_MS;
use crate::claim_receipt::timeliness::FULL_PPM;
use crate::claim_receipt::{rfc3339_seconds, ClaimReceipt, TimelinessEntry};
use crate::holdings::{Holdings, RecordVerdict};
use crate::jwp_delivery::{GroupReceipt, ObjectReceipt};

/// §7.5.1: no plan, one planned from `from_ms`, instants that do not strictly increase, or an
/// object wanted twice by one plan.
pub const PLANS_INVALID: &str = "plans_invalid";
/// §7.5.1: an arrival past the last instant, arrivals whose instants decrease, or two of one
/// object.
pub const ARRIVALS_INVALID: &str = "arrivals_invalid";

/// §7.5.1: the claims' instants are not one for each claim, one is past the last instant, a claim
/// is given twice, or a claim that passes jwp-receipts §3.1 steps 1 to 4 is held before it was signed.
pub const CLAIMS_INVALID: &str = "claims_invalid";
/// §7.5.1: the records' instants are not one for each record, one is past the last instant, a record
/// is given twice, or a record that reads is held before its `as_of_ms`.
pub const RECORDS_INVALID: &str = "records_invalid";

/// §7.5.1: the groups' instants are not one for each group, one is past the last instant, a group
/// is given twice, a group that reads is held before it starts, or the object receipts' instants are
/// not one for each object receipt of each group, or past the last instant.
pub const GROUPS_INVALID: &str = "groups_invalid";

/// A claim or a record a plan did not read, as [`Run::read`] labels it.
pub const NOT_HELD: &str = "not-held";

/// The decisions of §7.5.3, in their order there.
pub const ACTIONS: &[&str] = &["fetch", "abandon", "present", "passed", "ended"];

/// An object: a group's link hash and a position.
type Object = ([u8; 32], u64);

/// One plan of a run (§7.5.1): the instant it is given at and planned from, the sink's sources
/// in its order, as [`super::sources`] reads them, and the wanted objects.
#[derive(Clone, Debug, PartialEq)]
pub struct RunPlan {
    pub at: At,
    pub sources: Vec<Value>,
    pub wanted: Vec<Object>,
}

/// An arrival (§7.5.1): the instant from which the sink first holds an object, on its clock.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Arrival {
    pub group: [u8; 32],
    pub index: u64,
    pub at_ms: u64,
}

/// What the sink holds of an object on a bearer instance, from an instant on its clock (§7.5.1).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SymbolsAt {
    pub at_ms: u64,
    pub symbols: SymbolsHeld,
}

/// The inputs of §7.5.1.
#[derive(Clone, Debug)]
pub struct RunInputs {
    /// §7.1's groups, each plan reading those the sink held by its instant.
    pub groups: Vec<GroupInput>,
    /// The instant from which the sink first holds each group, and `None` for one held from before
    /// the run. `None` for a run that states no instant.
    pub groups_at: Option<Vec<Option<u64>>>,
    /// The same for each object receipt of each group, read once its group is.
    pub receipts_at: Option<Vec<Vec<Option<u64>>>>,
    /// §7.1's claims and records, each plan reading those the sink held by its instant.
    pub claims: Vec<String>,
    pub records: Vec<Vec<u8>>,
    /// The instant from which the sink first holds each claim, on its clock, and `None` for one held
    /// from before the run. `None` for a run that states no instant.
    pub claims_at: Option<Vec<Option<u64>>>,
    /// The same for each record.
    pub records_at: Option<Vec<Option<u64>>>,
    pub loss: Option<(u64, u64)>,
    /// The plans, the first in force from its instant and each later one replacing it.
    pub plans: Vec<RunPlan>,
    /// The sink's arrivals, in the order of its clock.
    pub arrivals: Vec<Arrival>,
    /// What the sink holds of objects on bearer instances, in the order of its clock; nothing of
    /// any it does not name.
    pub symbols: Vec<SymbolsAt>,
}

/// A decision's action (§7.5.3).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Action {
    Fetch,
    Abandon,
    Present,
    Passed,
    Ended,
}

impl Action {
    pub fn label(self) -> &'static str {
        match self {
            Action::Fetch => "fetch",
            Action::Abandon => "abandon",
            Action::Present => "present",
            Action::Passed => "passed",
            Action::Ended => "ended",
        }
    }
}

/// One decision (§7.5.3).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Decision {
    pub at_ms: u64,
    /// The plan in force; for an `abandon`, the plan the fetch served.
    pub plan: usize,
    pub action: Action,
    pub group: [u8; 32],
    pub index: u64,
    /// For `fetch` and `abandon` the object the fetch takes, for `present` the one presented.
    pub object: Option<Which>,
    /// The rung's index in the ladder, where the object is a rung.
    pub rung: Option<usize>,
    /// For `fetch` and `abandon`, the source's index in that plan's list.
    pub source: Option<usize>,
    /// For `present`, the value presented.
    pub value_ppm: Option<u64>,
}

/// What one plan read (§7.5.3): each claim's label and each record's, by their positions in the
/// run's inputs, as §7.2 gives them over what the plan read, and [`NOT_HELD`] for the rest.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PlanRead {
    pub claims: Vec<String>,
    pub records: Vec<String>,
}

/// §7.5's run: each plan's choices (§7.3), what each read where the run states instants, then the
/// decisions in the order taken.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Run {
    pub plans: Vec<Vec<Result<Choice, &'static str>>>,
    pub read: Option<Vec<PlanRead>>,
    pub decisions: Vec<Decision>,
}

/// A fetch outstanding (§7.5.2).
#[derive(Clone, Debug)]
struct Fetch {
    object: Which,
    rung: Option<usize>,
    /// The object it takes.
    target: Object,
    source: Source,
    /// The plan it serves and the source's index in that plan's list.
    plan: usize,
    index: usize,
}

#[derive(Default)]
struct State {
    settled: bool,
    passed: bool,
    fetch: Option<Fetch>,
}

/// One plan derived: its sources, and the objects it runs in its wanted order, each with its
/// choice and the entry that stands for it.
struct Derived {
    at: u64,
    sources: Vec<Source>,
    run: Vec<(Object, Choice, TimelinessEntry)>,
}

/// §7.5.2: two sources are the same where both name one bearer instance or one holder's key, or
/// both are the origin, whatever their `fetch_ms`.
fn same(a: &Source, b: &Source) -> bool {
    match (a, b) {
        (Source::Bearer(x), Source::Bearer(y)) => x == y,
        (Source::Holder { key: x, .. }, Source::Holder { key: y, .. }) => x == y,
        (Source::Origin { .. }, Source::Origin { .. }) => true,
        _ => false,
    }
}

/// The object a choice takes: the primary, or the rung it names.
fn target(object: &Object, entry: &TimelinessEntry, which: Which, rung: Option<usize>) -> Object {
    match (which, rung) {
        (Which::Fallback, Some(r)) => {
            let f = entry.fallback.rungs()[r];
            (f.group, f.leaf_index)
        }
        _ => *object,
    }
}

/// §7.5.1 checks 4 to 6 for one list: one entry for each item, none past the last instant, no
/// item twice, and none held in a whole second before the one `since` reads from it.
fn instants_formed<T: Ord>(items: &[T], at: &[Option<u64>], since: impl Fn(&T) -> Option<i64>) -> bool {
    at.len() == items.len()
        && at.iter().flatten().all(|&a| a <= LAST_INSTANT_MS)
        && items.iter().collect::<BTreeSet<_>>().len() == items.len()
        && items.iter().zip(at).all(|(x, a)| match (a, since(x)) {
            (Some(a), Some(s)) => (*a / 1000) as i64 >= s,
            _ => true,
        })
}

/// Whether what `at` dates at position `i` is held by `t`: its instant is at most `t`, or it states
/// none, and so was held from before the run.
fn held_by(at: &Option<Vec<Option<u64>>>, i: usize, t: u64) -> bool {
    at.as_ref().and_then(|v| v[i]).is_none_or(|a| a <= t)
}

/// What a plan at `t` reads of `items`, in the order given.
fn read_at<T: Clone>(items: &[T], at: &Option<Vec<Option<u64>>>, t: u64) -> Vec<T> {
    items.iter().enumerate().filter(|(i, _)| held_by(at, *i, t)).map(|(_, x)| x.clone()).collect()
}

/// §7.5.1 check 6, the object receipts' instants: one list for each group and one entry for each of
/// its object receipts, none past the last instant.
fn receipts_formed(groups: &[GroupInput], at: &[Vec<Option<u64>>]) -> bool {
    at.len() == groups.len()
        && groups.iter().zip(at).all(|(g, a)| a.len() == g.objects.len() && a.iter().flatten().all(|&t| t <= LAST_INSTANT_MS))
}

/// One object on one bearer instance: the `bearer_id`, the group's link hash and the position.
type OnInstance<'a> = (&'a str, [u8; 32], u64);

/// §7.5.1 check 7: instants at most the last instant, in order; each statement formed as §7.1 has
/// it; and of one object on one instance, each statement at a later instant than the one before it,
/// of as many blocks, and with no number below that one's.
fn symbols_trace_formed(trace: &[SymbolsAt]) -> bool {
    let mut last: BTreeMap<OnInstance<'_>, (u64, &[u64])> = BTreeMap::new();
    let mut previous = 0;
    for s in trace {
        let y = &s.symbols;
        if s.at_ms > LAST_INSTANT_MS || s.at_ms < previous || !y.formed() {
            return false;
        }
        previous = s.at_ms;
        let key = (y.bearer_id.as_str(), y.group, y.index);
        if let Some((at, held)) = last.get(&key)
            && (*at == s.at_ms || held.len() != y.held.len() || held.iter().zip(&y.held).any(|(a, b)| b < a))
        {
            return false;
        }
        last.insert(key, (s.at_ms, &y.held));
    }
    true
}

/// What the sink holds at `t` (§7.5.1): of each object on each instance, the last statement at or
/// before `t`.
fn symbols_read(trace: &[SymbolsAt], t: u64) -> Vec<SymbolsHeld> {
    let mut read: BTreeMap<OnInstance<'_>, &SymbolsHeld> = BTreeMap::new();
    for s in trace.iter().filter(|s| s.at_ms <= t) {
        read.insert((s.symbols.bearer_id.as_str(), s.symbols.group, s.symbols.index), &s.symbols);
    }
    read.into_values().cloned().collect()
}

/// §7.5.1: the groups a plan at `t` reads, in the order given. In a run that dates them, a group held
/// by `t`, with the object receipts held by `t`, each envelope only from its object's arrival, or
/// from before the first plan where no arrival names it. A run that dates none reads them whole.
fn groups_read(inp: &RunInputs, t: u64) -> Vec<GroupInput> {
    if inp.groups_at.is_none() && inp.receipts_at.is_none() {
        return inp.groups.clone();
    }
    let keep = |(i, g): (usize, &GroupInput)| {
        if !held_by(&inp.groups_at, i, t) {
            return None;
        }
        let link = GroupReceipt::from_json(&g.group).ok().map(|r| r.link_hash());
        let objects = g
            .objects
            .iter()
            .enumerate()
            .filter(|(k, _)| inp.receipts_at.as_ref().and_then(|at| at[i][*k]).is_none_or(|a| a <= t))
            .map(|(_, o)| {
                let id = ObjectReceipt::from_json(&o.receipt).ok().map(|r| r.object_id);
                let arrival = inp.arrivals.iter().find(|a| Some(a.group) == link && Some(a.index) == id);
                let held = arrival.is_none_or(|a| a.at_ms <= t);
                HeldObject { receipt: o.receipt.clone(), envelope: o.envelope.clone().filter(|_| held) }
            })
            .collect();
        Some(GroupInput { group: g.group.clone(), objects })
    };
    inp.groups.iter().enumerate().filter_map(keep).collect()
}

/// The labels of §7.5.3 for `n` items: each held by `t` takes the next of `labels`, in order, and
/// the rest are [`NOT_HELD`].
fn labelled(n: usize, at: &Option<Vec<Option<u64>>>, t: u64, labels: Vec<String>) -> Vec<String> {
    let mut labels = labels.into_iter();
    (0..n).map(|i| if held_by(at, i, t) { labels.next().expect("a label for each item read") } else { NOT_HELD.to_owned() }).collect()
}

/// §7.5.1: the checks, in their order.
fn check(inp: &RunInputs) -> Result<(), Refusal> {
    let instant = |a: At| match a {
        At::Instant(t) => Some(t),
        At::FromMs => None,
    };
    let formed = !inp.plans.is_empty()
        && inp.plans.iter().all(|p| instant(p.at).is_some())
        && inp.plans.windows(2).all(|w| instant(w[0].at) < instant(w[1].at))
        // An entry §7.1 refuses names no object: that is check 2's, not this one's.
        && inp.plans.iter().all(|p| {
            let named: Vec<&Object> = p.wanted.iter().filter(|o| position_ok(o.1)).collect();
            named.iter().collect::<BTreeSet<_>>().len() == named.len()
        });
    if !formed {
        return Err(Refusal::Input(PLANS_INVALID));
    }
    for p in &inp.plans {
        checked(inp.loss, p.at, &p.sources)?;
        if !wanted_formed(&p.wanted) {
            return Err(Refusal::Input(WANTED_INVALID));
        }
    }
    let arrivals_formed = inp.arrivals.iter().all(|a| a.at_ms <= LAST_INSTANT_MS)
        && inp.arrivals.windows(2).all(|w| w[0].at_ms <= w[1].at_ms)
        && inp.arrivals.iter().map(|a| (a.group, a.index)).collect::<BTreeSet<_>>().len() == inp.arrivals.len();
    if !arrivals_formed {
        return Err(Refusal::Input(ARRIVALS_INVALID));
    }
    // A claim that fails steps 1 to 4 is not known to be its signer's, so its `issued_at` is no one's.
    let signed = |wire: &String| {
        let c = ClaimReceipt::from_json(wire).ok().filter(|c| c.verify().is_ok())?;
        rfc3339_seconds(&c.issued_at)
    };
    if inp.claims_at.as_ref().is_some_and(|at| !instants_formed(&inp.claims, at, signed)) {
        return Err(Refusal::Input(CLAIMS_INVALID));
    }
    let stated = |bytes: &Vec<u8>| Holdings::read(bytes).ok().map(|h| (h.as_of_ms / 1000) as i64);
    if inp.records_at.as_ref().is_some_and(|at| !instants_formed(&inp.records, at, stated)) {
        return Err(Refusal::Input(RECORDS_INVALID));
    }
    // A group that fails jwp-receipts §3 steps 1 and 2 starts at no instant it can be held to.
    let starts = |g: &GroupInput| {
        let r = GroupReceipt::from_json(&g.group).ok().filter(|r| r.verify().is_ok())?;
        rfc3339_seconds(&r.group_start)
    };
    let keyed: Vec<(String, Option<i64>)> = inp.groups.iter().map(|g| (g.group.to_string(), starts(g))).collect();
    if inp.groups_at.as_ref().is_some_and(|at| !instants_formed(&keyed, at, |k| k.1))
        || inp.receipts_at.as_ref().is_some_and(|at| !receipts_formed(&inp.groups, at))
    {
        return Err(Refusal::Input(GROUPS_INVALID));
    }
    if !symbols_trace_formed(&inp.symbols) {
        return Err(Refusal::Input(SYMBOLS_INVALID));
    }
    Ok(())
}

/// §7.5: the run of a sink's plans against its arrivals.
pub fn run(inp: &RunInputs) -> Result<Run, Refusal> {
    check(inp)?;
    // Each plan derived by §7.2 and §7.3 from what the sink held by its instant.
    let mut plans: Vec<Vec<Result<Choice, &'static str>>> = Vec::with_capacity(inp.plans.len());
    let mut derived: Vec<Derived> = Vec::with_capacity(inp.plans.len());
    let mut read: Vec<PlanRead> = Vec::with_capacity(inp.plans.len());
    for p in &inp.plans {
        let At::Instant(at) = p.at else { unreachable!("checked: every plan is at an instant") };
        let pi = PlanInputs {
            groups: groups_read(inp, at),
            symbols: symbols_read(&inp.symbols, at),
            claims: read_at(&inp.claims, &inp.claims_at, at),
            records: read_at(&inp.records, &inp.records_at, at),
            loss: inp.loss,
            at: p.at,
            sources: p.sources.clone(),
            wanted: p.wanted.clone(),
        };
        let (verdict, entries) = planned(&pi)?;
        let sources = checked(inp.loss, p.at, &p.sources)?;
        read.push(PlanRead {
            claims: labelled(inp.claims.len(), &inp.claims_at, at, verdict.claims.iter().map(ClaimVerdict::label).collect()),
            records: labelled(inp.records.len(), &inp.records_at, at, verdict.records.iter().map(RecordVerdict::label).collect()),
        });
        let run = p
            .wanted
            .iter()
            .zip(&verdict.choices)
            .zip(entries)
            .filter_map(|((o, c), e)| Some((*o, *c.as_ref().ok()?, e?)))
            .collect();
        derived.push(Derived { at, sources, run });
        plans.push(verdict.choices);
    }

    // The instants the run reads: each plan's, each arrival's, and each deadline and end.
    let mut instants: BTreeSet<u64> = derived.iter().map(|d| d.at).collect();
    instants.extend(inp.arrivals.iter().map(|a| a.at_ms));
    for d in &derived {
        for (_, _, e) in &d.run {
            instants.insert(e.by_ms);
            instants.insert(e.end_ms());
        }
    }

    let mut held: BTreeMap<Object, u64> = BTreeMap::new();
    let mut state: BTreeMap<Object, State> = BTreeMap::new();
    let mut in_force: Option<usize> = None;
    let mut out: Vec<Decision> = Vec::new();
    let decide = |out: &mut Vec<Decision>, at_ms, plan, action, o: &Object, object, rung, source, value_ppm| {
        out.push(Decision { at_ms, plan, action, group: o.0, index: o.1, object, rung, source, value_ppm });
    };
    let abandon = |out: &mut Vec<Decision>, n: u64, o: &Object, f: &Fetch| {
        out.push(Decision {
            at_ms: n,
            plan: f.plan,
            action: Action::Abandon,
            group: o.0,
            index: o.1,
            object: Some(f.object),
            rung: f.rung,
            source: Some(f.index),
            value_ppm: None,
        });
    };

    for n in instants {
        // Step 1: arrivals.
        for a in inp.arrivals.iter().filter(|a| a.at_ms == n) {
            let o = (a.group, a.index);
            held.insert(o, n);
            for st in state.values_mut() {
                if st.fetch.as_ref().is_some_and(|f| f.target == o) {
                    st.fetch = None;
                }
            }
        }

        // Step 2: a plan given at n replaces the one in force.
        if let Some(k) = derived.iter().position(|d| d.at == n) {
            let new = &derived[k];
            let continues = |o: &Object, f: &Fetch| {
                new.run.iter().any(|(w, c, e)| {
                    w == o
                        && match (c.object, c.source) {
                            (Some(which), Some(si)) => target(w, e, which, c.rung) == f.target && same(&new.sources[si], &f.source),
                            _ => false,
                        }
                })
            };
            if let Some(old) = in_force {
                for (o, _, _) in &derived[old].run {
                    let st = state.entry(*o).or_default();
                    if st.settled {
                        continue;
                    }
                    if let Some(f) = st.fetch.take_if(|f| !continues(o, f)) {
                        abandon(&mut out, n, o, &f);
                    }
                }
            }
            in_force = Some(k);
            for (o, c, e) in &new.run {
                let st = state.entry(*o).or_default();
                let (Some(which), Some(si)) = (c.object, c.source) else { continue };
                if st.settled {
                    continue;
                }
                let t = target(o, e, which, c.rung);
                if held.contains_key(&t) {
                    continue;
                }
                if let Some(f) = st.fetch.as_mut()
                    && f.target == t
                    && same(&f.source, &new.sources[si])
                {
                    // Continued: it serves the new plan from now on.
                    (f.plan, f.index) = (k, si);
                    continue;
                }
                st.fetch = Some(Fetch { object: which, rung: c.rung, target: t, source: new.sources[si].clone(), plan: k, index: si });
                decide(&mut out, n, k, Action::Fetch, o, Some(which), c.rung, Some(si), None);
            }
        }

        let Some(k) = in_force else { continue };
        // Step 3: deadlines, by §5.3 with what the sink holds by `by_ms`.
        for (o, _, e) in &derived[k].run {
            let st = state.entry(*o).or_default();
            if st.settled || st.passed || e.by_ms > n {
                continue;
            }
            let by = |t: &Object| held.get(t).is_some_and(|&a| a <= e.by_ms);
            if by(o) {
                st.settled = true;
                decide(&mut out, n, k, Action::Present, o, Some(Which::Primary), None, None, Some(FULL_PPM));
            } else if let Some((r, f)) = e.fallback.rungs().iter().enumerate().find(|(_, f)| by(&(f.group, f.leaf_index))) {
                st.settled = true;
                decide(&mut out, n, k, Action::Present, o, Some(Which::Fallback), Some(r), None, Some(f.value_ppm));
            } else {
                st.passed = true;
                decide(&mut out, n, k, Action::Passed, o, None, None, None, None);
            }
            // After a `present`, any fetch; after `passed`, a rung's.
            let settled = st.settled;
            if let Some(f) = st.fetch.take_if(|f| settled || f.object == Which::Fallback) {
                abandon(&mut out, n, o, &f);
            }
        }
        // Step 4: late primaries.
        for (o, _, e) in &derived[k].run {
            let st = state.entry(*o).or_default();
            if !st.passed || st.settled {
                continue;
            }
            if let Some(&a) = held.get(o) {
                let v = e.value_at(a);
                if v >= 1 {
                    st.settled = true;
                    decide(&mut out, n, k, Action::Present, o, Some(Which::Primary), None, None, Some(v));
                }
            }
        }
        // Step 5: ends.
        for (o, _, e) in &derived[k].run {
            let st = state.entry(*o).or_default();
            if !st.passed || st.settled || e.end_ms() > n {
                continue;
            }
            if let Some(f) = st.fetch.take() {
                abandon(&mut out, n, o, &f);
            }
            st.settled = true;
            decide(&mut out, n, k, Action::Ended, o, None, None, None, None);
        }
    }
    let dated = inp.claims_at.is_some() || inp.records_at.is_some() || inp.groups_at.is_some() || inp.receipts_at.is_some();
    Ok(Run { plans, read: dated.then_some(read), decisions: out })
}

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

    #[test]
    fn sources_are_the_same_by_what_they_name() {
        let key = [3; 32];
        assert!(same(&Source::Origin { fetch_ms: 1 }, &Source::Origin { fetch_ms: 9 }));
        assert!(same(&Source::Holder { key, fetch_ms: 1 }, &Source::Holder { key, fetch_ms: 2 }));
        assert!(!same(&Source::Holder { key, fetch_ms: 1 }, &Source::Holder { key: [4; 32], fetch_ms: 1 }));
        assert!(same(&Source::Bearer("urn:example:bearer:b1".into()), &Source::Bearer("urn:example:bearer:b1".into())));
        assert!(!same(&Source::Bearer("urn:example:bearer:b1".into()), &Source::Origin { fetch_ms: 1 }));
    }

    #[test]
    fn the_inputs_are_checked_in_order() {
        let base = || RunInputs {
            groups: Vec::new(),
            claims: Vec::new(),
            records: Vec::new(),
            claims_at: None,
            records_at: None,
            groups_at: None,
            receipts_at: None,
            symbols: Vec::new(),
            loss: None,
            plans: vec![RunPlan { at: At::Instant(5), sources: vec![serde_json::json!({"kind": "origin", "fetch_ms": 1})], wanted: vec![([1; 32], 0)] }],
            arrivals: vec![Arrival { group: [1; 32], index: 0, at_ms: 7 }],
        };
        assert_eq!(check(&base()), Ok(()));
        let with = |f: &dyn Fn(&mut RunInputs)| {
            let mut i = base();
            f(&mut i);
            check(&i)
        };
        let plans = Err(Refusal::Input(PLANS_INVALID));
        let arrivals = Err(Refusal::Input(ARRIVALS_INVALID));
        assert_eq!(with(&|i| i.plans.clear()), plans);
        assert_eq!(with(&|i| i.plans[0].at = At::FromMs), plans);
        assert_eq!(with(&|i| i.plans.push(i.plans[0].clone())), plans, "two plans at one instant");
        assert_eq!(with(&|i| i.plans[0].wanted.push(([1; 32], 0))), plans);
        assert_eq!(with(&|i| i.plans[0].sources.clear()), Err(Refusal::Input(super::super::SOURCES_INVALID)));
        assert_eq!(
            with(&|i| {
                i.plans[0].sources.clear();
                i.arrivals.push(i.arrivals[0]);
            }),
            Err(Refusal::Input(super::super::SOURCES_INVALID)),
            "a plan's reason before the arrivals'"
        );
        assert_eq!(with(&|i| i.arrivals[0].at_ms = LAST_INSTANT_MS + 1), arrivals);
        assert_eq!(with(&|i| i.arrivals.push(i.arrivals[0])), arrivals, "one object twice");
        assert_eq!(with(&|i| i.arrivals.insert(0, Arrival { group: [2; 32], index: 0, at_ms: 8 })), arrivals, "instants that decrease");
        assert_eq!(with(&|i| i.arrivals.insert(0, Arrival { group: [2; 32], index: 0, at_ms: 7 })), Ok(()), "one instant twice");
        let claims = Err(Refusal::Input(CLAIMS_INVALID));
        let records = Err(Refusal::Input(RECORDS_INVALID));
        let given = |i: &mut RunInputs| {
            i.claims = vec!["one".into(), "two".into()];
            i.records = vec![vec![1], vec![2]];
        };
        assert_eq!(with(&|i| { given(i); i.claims_at = Some(vec![None, Some(9)]); i.records_at = Some(vec![Some(9), None]); }), Ok(()), "what does not read is not compared");
        assert_eq!(with(&|i| { given(i); i.claims_at = Some(vec![None]); }), claims, "not one for each claim");
        assert_eq!(with(&|i| { given(i); i.claims_at = Some(vec![None, Some(LAST_INSTANT_MS + 1)]); }), claims);
        assert_eq!(with(&|i| { given(i); i.claims[1] = "one".into(); i.claims_at = Some(vec![None, None]); }), claims, "one claim twice");
        assert_eq!(with(&|i| { given(i); i.records_at = Some(vec![None]); }), records, "not one for each record");
        assert_eq!(with(&|i| { given(i); i.records_at = Some(vec![None, Some(LAST_INSTANT_MS + 1)]); }), records);
        assert_eq!(with(&|i| { given(i); i.records[1] = vec![1]; i.records_at = Some(vec![None, None]); }), records, "one record twice");
        assert_eq!(with(&|i| { given(i); i.claims_at = Some(vec![None]); i.records_at = Some(vec![None]); }), claims, "the claims' reason before the records'");
        assert_eq!(with(&|i| { given(i); i.claims[1] = "one".into(); }), Ok(()), "a run that states no instant takes its claims as they come");
    }

    #[test]
    fn a_plan_reads_what_was_held_by_its_instant() {
        let at = Some(vec![None, Some(10), Some(20)]);
        assert_eq!(read_at(&["a", "b", "c"], &at, 9), vec!["a"], "no instant: held from before the run");
        assert_eq!(read_at(&["a", "b", "c"], &at, 10), vec!["a", "b"], "held at the plan's own instant");
        assert_eq!(read_at(&["a", "b", "c"], &at, 20), vec!["a", "b", "c"]);
        assert_eq!(read_at(&["a", "b"], &None, 0), vec!["a", "b"], "a run that states no instant");
        let labels = labelled(3, &at, 10, vec!["x".into(), "y".into()]);
        assert_eq!(labels, vec!["x".to_owned(), "y".to_owned(), NOT_HELD.to_owned()]);
    }

    fn said(at_ms: u64, id: &str, held: &[u64]) -> SymbolsAt {
        SymbolsAt { at_ms, symbols: SymbolsHeld { bearer_id: id.into(), group: [1; 32], index: 0, held: held.to_vec() } }
    }

    #[test]
    fn what_the_sink_holds_is_a_trace_that_never_falls() {
        let b = "urn:example:b";
        assert!(symbols_trace_formed(&[]));
        assert!(symbols_trace_formed(&[said(5, b, &[1, 2]), said(6, b, &[1, 3])]));
        assert!(symbols_trace_formed(&[said(5, b, &[1]), said(5, "urn:example:c", &[1])]), "one instant on two instances");
        assert!(!symbols_trace_formed(&[said(5, b, &[2]), said(6, b, &[1])]), "a symbol unlearned");
        assert!(!symbols_trace_formed(&[said(5, b, &[1]), said(6, b, &[1, 1])]), "another count of blocks");
        assert!(!symbols_trace_formed(&[said(5, b, &[1]), said(5, b, &[2])]), "one object at one instant, twice");
        assert!(!symbols_trace_formed(&[said(6, b, &[1]), said(5, "urn:example:c", &[1])]), "instants that decrease");
        assert!(!symbols_trace_formed(&[said(LAST_INSTANT_MS + 1, b, &[1])]));
        assert!(!symbols_trace_formed(&[said(5, b, &[])]), "no number");
        assert!(!symbols_trace_formed(&[said(5, "b", &[1])]), "not a URI");
        assert!(!symbols_trace_formed(&[said(5, b, &[1 << 53])]), "past the greatest integer");
        assert!(symbols_trace_formed(&[said(5, b, &[(1 << 53) - 1])]));
    }

    #[test]
    fn a_plan_counts_the_last_statement_by_its_instant() {
        let b = "urn:example:b";
        let trace = [said(5, b, &[1]), said(10, b, &[2])];
        assert!(symbols_read(&trace, 4).is_empty());
        assert_eq!(symbols_read(&trace, 5)[0].held, vec![1]);
        assert_eq!(symbols_read(&trace, 9)[0].held, vec![1]);
        assert_eq!(symbols_read(&trace, 10)[0].held, vec![2]);
    }

    #[test]
    fn an_instant_is_read_in_whole_seconds() {
        let plain = |_: &u8| Some(5);
        assert!(instants_formed(&[1u8], &[Some(5_000)], plain));
        assert!(instants_formed(&[1u8], &[Some(5_999)], plain));
        assert!(!instants_formed(&[1u8], &[Some(4_999)], plain));
        assert!(instants_formed(&[1u8], &[None], plain), "no instant: nothing to compare");
    }
}
