//! The live schedule of a stage set (`extensions/staged-delivery.md` §8): when each
//! unit is presented, whether a layer arrived in time to be applied to it, how many
//! layers a sink can fetch before a unit is due, and what a sink presents for a trace
//! of arrivals. Pure integer functions of their inputs, so the corpus's `schedule.json`
//! is replayed exactly by the Rust and the Python verifiers.

/// The timing of a stage set's units: the unit at position `u` on the clock is
/// presented at `t0_ms + u·unit_ms + target_latency_ms`. A base's unit `i` sits at
/// position `first_unit + i` ([`super::BaseInfo::first_unit`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct UnitClock {
    pub t0_ms: u64,
    pub unit_ms: u64,
    pub target_latency_ms: u64,
}

/// One layer a publisher offers: its index, its lead (the time before a unit's
/// presentation by which it must arrive), and its bytes as delivered.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct LayerPlan {
    pub layer: u8,
    pub lead_ms: u64,
    pub bytes: u64,
}

/// When the unit at clock position `unit` is presented (saturating at `u64::MAX`).
pub fn presentation_ms(c: &UnitClock, unit: u64) -> u64 {
    c.t0_ms.saturating_add(unit.saturating_mul(c.unit_ms)).saturating_add(c.target_latency_ms)
}

/// Whether a layer that arrived at `arrival_ms` with lead `lead_ms` is in time for
/// the unit at clock position `unit`: it arrived at least `lead_ms` before the unit's
/// presentation (§8).
pub fn on_time(c: &UnitClock, unit: u64, arrival_ms: u64, lead_ms: u64) -> bool {
    arrival_ms.checked_add(lead_ms).is_some_and(|t| t <= presentation_ms(c, unit))
}

/// The milliseconds `bytes` take at `throughput_bps` bits per second, rounded up;
/// `u64::MAX` at a throughput of 0.
pub fn transfer_ms(bytes: u64, throughput_bps: u64) -> u64 {
    if throughput_bps == 0 {
        return u64::MAX;
    }
    let bits = (bytes as u128) * 8 * 1000;
    u64::try_from(bits.div_ceil(throughput_bps as u128)).unwrap_or(u64::MAX)
}

/// How many layers a sink fetching from `now_ms` at `throughput_bps` can apply to a
/// unit due at `due_ms`: the largest `k` such that, for every `j ≤ k`, layers `1..=j`
/// fetched in order arrive at least `plan[j].lead_ms` before `due_ms`. `plan` is in
/// layer order. Integer arithmetic only, so the answer is the same everywhere.
pub fn choose_layers(throughput_bps: u64, now_ms: u64, due_ms: u64, plan: &[LayerPlan]) -> u8 {
    let mut total: u64 = 0;
    let mut k: u8 = 0;
    for p in plan {
        total = total.saturating_add(p.bytes);
        let arrives = now_ms.saturating_add(transfer_ms(total, throughput_bps));
        if arrives.checked_add(p.lead_ms).is_none_or(|t| t > due_ms) {
            break;
        }
        k = k.saturating_add(1);
    }
    k
}

/// What a sink presents for one unit of a trace: when, and at which stage.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Presented {
    pub at_ms: u64,
    pub stage: u8,
    /// The stage the layers the trace holds for this unit would have reached had
    /// each arrived in time.
    pub expected: u8,
}

/// What a sink presents for a trace (§8). `base_ms[u]` is when unit `u`'s base
/// arrived; `layers[l][u]`, when layer `l + 1` arrived for unit `u` (`None`: it
/// never did); `lead_ms[l]`, layer `l + 1`'s lead. A unit is presented at its
/// presentation time, or when its base arrives if that is later: the sink never waits
/// for a layer. Its stage is the longest run of layers `1, 2, …` each of which arrived
/// in time; a layer that arrived late is not applied, and no layer above it is. Its
/// expected stage is the longest run of layers the trace holds for it.
pub fn present(c: &UnitClock, base_ms: &[u64], layers: &[Vec<Option<u64>>], lead_ms: &[u64]) -> Vec<Presented> {
    base_ms
        .iter()
        .enumerate()
        .map(|(u, &b)| {
            let unit = u as u64;
            let at_ms = presentation_ms(c, unit).max(b);
            let mut stage = 0u8;
            let mut expected = 0u8;
            let mut in_time = true;
            for (l, arrivals) in layers.iter().enumerate() {
                let Some(Some(t)) = arrivals.get(u) else { break };
                expected = expected.saturating_add(1);
                in_time = in_time && on_time(c, unit, *t, lead_ms.get(l).copied().unwrap_or(0));
                if in_time {
                    stage = stage.saturating_add(1);
                }
            }
            Presented { at_ms, stage, expected }
        })
        .collect()
}

/// The units a trace presents below the stage their layers would have reached on
/// time (§8): the count a sink keeps.
pub fn deadline_misses(p: &[Presented]) -> u64 {
    p.iter().filter(|x| x.stage < x.expected).count() as u64
}
