//! Progress reporting for pipeline runs: a full-screen dashboard when
//! stdout is a terminal, or line-oriented output when it is not (or when
//! `--plain` asks for it).
//!
//! Both reporters consume the pipeline's event channel until it
//! disconnects (which happens exactly when the pipeline finishes) and
//! let the user request suspension — with a key in the dashboard, or
//! SIGINT/SIGTERM in either mode.
//!
//! The dashboard answers three questions at once: what the run has
//! accomplished (the totals the pipeline itself keeps), what its workers
//! are doing this instant (the stages table), and what is queued ahead
//! of them (the table's waiting column for how deep each queue is now,
//! the backlog pane for which way it is going, and the pending pane for
//! the paths a run has yet to visit). Completed
//! items scroll past beside those, and skips — which are reports rather
//! than problems, and can be very numerous — are reduced to the most
//! recent one alongside their count.

use std::{
    collections::{HashMap, VecDeque},
    io::Write as _,
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::{Duration, Instant},
};

use crossbeam_channel::Receiver;
use crossterm::event::{Event as TermEvent, KeyCode, KeyModifiers};
use ratatui::{
    Frame,
    layout::{Constraint, Layout},
    style::{Modifier, Style},
    text::Line,
    widgets::{Block, Borders, Cell, List, ListItem, Paragraph, Row, Table},
};

use beeping::pipeline::{EarlyFinish, Event, Session, SkipReason, Stage, StoreLimit};

/// History kept for the activity pane; the visible tail is cut to the
/// pane's actual height at render time.
const HISTORY_LINES: usize = 128;

/// How much run time separates two backlog samples. The pipeline
/// reports several times a second, which would fill the backlog pane
/// with a few seconds of noise; a couple of seconds a column gives it
/// several minutes, which is the span over which a queue meaningfully
/// grows or drains.
const BACKLOG_SAMPLE: Duration = Duration::from_secs(2);

/// Backlog samples kept per stage, comfortably more than the widest
/// pane, so that history survives a narrow window being widened.
const BACKLOG_HISTORY: usize = 512;

/// The width the activity log gives its verb, so that the paths beside
/// it line up.
const VERB_WIDTH: u16 = 8;

/// How much recent history throughput is averaged over. Long enough to
/// ride out one slow file, short enough to still mean "right now".
const RATE_WINDOW: Duration = Duration::from_secs(5);

/// The shortest span of samples that yields a throughput figure; below
/// it the quotient is more noise than measurement.
const RATE_MIN_SPAN: Duration = Duration::from_millis(500);

/// One line of the activity log.
enum Activity {
    /// Something that happened to a path. The verb and the path are
    /// kept apart rather than formatted when the event arrives, because
    /// how much room there is for the path is not known until the pane
    /// is drawn — and what must survive the cut is its tail.
    Item { verb: &'static str, path: String },
    /// Something the run has to say about itself, which is prose rather
    /// than a path and is left whole.
    Note(String),
}

impl Activity {
    fn render(&self, width: u16) -> String {
        match self {
            Activity::Item { verb, path } => format!(
                "{verb:<label$} {}",
                fit_path(path, width.saturating_sub(VERB_WIDTH + 1)),
                label = VERB_WIDTH as usize
            ),
            Activity::Note(note) => note.clone(),
        }
    }
}

/// One item in a worker's hands.
struct InHand {
    /// When it was taken up, so the newest and the oldest can be told
    /// apart.
    sequence: u64,
    path: String,
    /// How far through it the worker has got, where the stage knows and
    /// the item is big enough for it to matter.
    progress: Option<(u64, u64)>,
}

impl InHand {
    /// The item as a table cell: how far through it the worker is, then
    /// as much of its path as the column can hold. The percentage keeps
    /// its full width — it is the part that changes, and the part that
    /// says whether anything is happening at all.
    fn describe(&self, width: u16) -> String {
        match self.progress {
            Some((done, total)) if total > 0 => {
                let percent = (done.saturating_mul(100) / total).min(100);
                let shown = format!("{percent:>3}% ");

                format!(
                    "{shown}{}",
                    fit_path(&self.path, width.saturating_sub(shown.len() as u16))
                )
            }
            _ => fit_path(&self.path, width),
        }
    }
}

/// What one stage of the pipeline is doing, and how much is queued for
/// it.
struct StageView {
    stage: Stage,
    /// How many workers the stage runs, or `None` for the stages whose
    /// work finalization does rather than a pool.
    threads: Option<usize>,
    /// The items in workers' hands, by worker index.
    busy: HashMap<usize, InHand>,
    waiting: u64,
    /// How deep the queue has been, oldest sample first: what the
    /// backlog pane draws.
    history: VecDeque<u64>,
    /// The head of the queue by name, in the order it will be taken up,
    /// or `None` for a stage whose backlog is not a queue of paths.
    upcoming: Option<Vec<String>>,
}

impl StageView {
    fn new(stage: Stage) -> StageView {
        StageView {
            stage,
            threads: None,
            busy: HashMap::new(),
            waiting: 0,
            history: VecDeque::new(),
            upcoming: None,
        }
    }

    /// Takes one backlog sample, dropping the oldest once the history is
    /// longer than any pane could use.
    fn record(&mut self, waiting: u64) {
        self.history.push_back(waiting);

        if self.history.len() > BACKLOG_HISTORY {
            self.history.pop_front();
        }
    }

    /// The item the stage took up most recently and still holds: where
    /// the run has got to.
    fn newest(&self) -> Option<&InHand> {
        self.busy.values().max_by_key(|item| item.sequence)
    }

    /// The item that has been in the stage's hands longest: the one
    /// holding the stage up, which the newest item hides whenever the
    /// other workers are churning past it. Named for what it measures —
    /// how long it has been held — rather than for the slowness that is
    /// usually, but not always, the reason.
    ///
    /// Both are drawn from the items actually in hand, so neither ever
    /// names something already finished, and a stage with one item names
    /// it in both columns.
    fn oldest(&self) -> Option<&InHand> {
        self.busy.values().min_by_key(|item| item.sequence)
    }

    /// The "busy" column: workers occupied out of workers available.
    fn occupancy(&self) -> String {
        match self.threads {
            Some(threads) => format!("{}/{}", self.busy.len(), threads),
            None => "—".to_string(),
        }
    }
}

/// A throughput estimate in bytes per second, over the last few seconds
/// of totals samples.
///
/// Both quantities are the pipeline's own — the time it has spent
/// running and the bytes it has moved — so the estimate does not depend
/// on how often the display redraws.
#[derive(Default)]
struct Rate {
    samples: VecDeque<(Duration, u64)>,
}

impl Rate {
    fn observe(&mut self, elapsed: Duration, bytes: u64) {
        // Samples only ever move forward within a run; anything else is
        // a fresh pipeline reporting into the same display
        if let Some((newest, _)) = self.samples.back()
            && elapsed < *newest
        {
            self.samples.clear();
        }

        self.samples.push_back((elapsed, bytes));

        // Keep the shortest run of samples that still spans the window
        while self.samples.len() > 2 && elapsed - self.samples[1].0 >= RATE_WINDOW {
            self.samples.pop_front();
        }
    }

    fn per_second(&self) -> Option<u64> {
        let (oldest, then) = *self.samples.front()?;
        let (newest, now) = *self.samples.back()?;
        let span = newest - oldest;

        if span < RATE_MIN_SPAN {
            return None;
        }

        Some((now.saturating_sub(then) as f64 / span.as_secs_f64()) as u64)
    }
}

/// What the run is doing, as far as its events have said.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
enum Phase {
    #[default]
    Working,
    /// Storing a manifest for everything backed up so far, so that the
    /// suspended run can be restored from.
    Checkpointing,
    /// All content is stored; the snapshot is being built.
    Finalizing,
}

/// Everything the reporters display, fed by pipeline events.
#[derive(Default)]
struct Dashboard {
    /// The run's totals, as the pipeline last reported them. It keeps
    /// these counts for its own durable state, so taking them from it
    /// rather than re-deriving them here keeps one set of numbers.
    totals: Session,
    phase: Phase,
    recent: VecDeque<Activity>,
    last_skip: Option<String>,
    /// Stage views in the order the pipeline first mentioned them, which
    /// is the order work flows through them.
    stages: Vec<StageView>,
    /// Throughput of finished content: the bytes of files stored or
    /// restored.
    content_rate: Rate,
    /// Throughput to and from the repository. Distinct from the content
    /// rate, and worth showing beside it: a run chewing through one huge
    /// file finishes no content while transferring steadily, and a
    /// re-backup of unchanged files does the reverse.
    transfer_rate: Rate,
    /// Bytes moved to or from the repository this session, accumulated
    /// as they are reported and sampled whenever new totals arrive.
    transferred: u64,
    /// Ticks up with every item picked up anywhere, so each stage can
    /// tell which of the items in its hands it took up last.
    sequence: u64,
    /// The pipeline's running time as of the last backlog sample, which
    /// is what paces the next one.
    backlog_sampled: Option<Duration>,
}

impl Dashboard {
    fn apply(&mut self, event: Event) {
        match event {
            Event::Scanned { path } => self.push_recent(Activity::Item {
                verb: "scanned",
                path: path.display().to_string(),
            }),
            Event::Stored { path, .. } => self.push_recent(Activity::Item {
                verb: "stored",
                path: path.display().to_string(),
            }),
            Event::Restored { path, .. } => self.push_recent(Activity::Item {
                verb: "restored",
                path: path.display().to_string(),
            }),
            Event::Skipped { path, reason } => {
                self.last_skip = Some(format!("{} ({})", path.display(), describe_skip(&reason)));
            }
            Event::Checkpointing => self.phase = Phase::Checkpointing,
            Event::Working => self.phase = Phase::Working,
            Event::FinishingEarly(reason) => self.push_recent(Activity::Note(format!(
                "finishing early: {}",
                describe_early_finish(reason)
            ))),
            Event::LimitNotApplied(limit) => {
                self.push_recent(Activity::Note(describe_unapplied_limit(limit)))
            }
            Event::Finalizing => self.phase = Phase::Finalizing,
            Event::Resumed(session) => {
                self.observe(session);
                self.push_recent(Activity::Note(
                    "resumed: totals include previous sessions".to_string(),
                ));
            }
            Event::Totals(session) => self.observe(session),
            Event::Staffing { stage, threads } => self.stage(stage).threads = Some(threads),
            Event::Began {
                stage,
                worker,
                path,
            } => {
                self.sequence += 1;
                let sequence = self.sequence;

                self.stage(stage).busy.insert(
                    worker,
                    InHand {
                        sequence,
                        path: path.display().to_string(),
                        progress: None,
                    },
                );
            }
            Event::Progressed {
                stage,
                worker,
                done,
                total,
            } => {
                if let Some(item) = self.stage(stage).busy.get_mut(&worker) {
                    item.progress = Some((done, total));
                }
            }
            Event::Idle { stage, worker } => {
                self.stage(stage).busy.remove(&worker);
            }
            Event::Backlog(stages) => {
                // Depths are taken as they arrive; history is sampled
                // against the pipeline's own running time, as the
                // throughput figures are, because the display is fed
                // far more often than the history needs a column.
                let elapsed = self.totals.elapsed;

                // Time that has gone backwards is a fresh pipeline
                // reporting into the same display, and its
                // predecessor's queues do not continue into this one
                if self.backlog_sampled.is_some_and(|last| elapsed < last) {
                    self.backlog_sampled = None;

                    for view in &mut self.stages {
                        view.history.clear();
                    }
                }

                let sampling = self
                    .backlog_sampled
                    .is_none_or(|last| elapsed - last >= BACKLOG_SAMPLE);

                if sampling {
                    self.backlog_sampled = Some(elapsed);
                }

                for backlog in stages {
                    let view = self.stage(backlog.stage);
                    view.waiting = backlog.waiting;
                    view.upcoming = backlog.upcoming.map(|paths| {
                        paths
                            .iter()
                            .map(|path| path.display().to_string())
                            .collect()
                    });

                    if sampling {
                        view.record(backlog.waiting);
                    }
                }
            }
            Event::Transferred { bytes } => self.transferred += bytes,
        }
    }

    /// The view of one stage, created on first mention so the table
    /// grows in pipeline order without either reporter knowing which
    /// stages a given pipeline has.
    fn stage(&mut self, stage: Stage) -> &mut StageView {
        let position = self
            .stages
            .iter()
            .position(|view| view.stage == stage)
            .unwrap_or_else(|| {
                self.stages.push(StageView::new(stage));
                self.stages.len() - 1
            });

        &mut self.stages[position]
    }

    /// Takes a fresh set of totals, and with them a sampling point for
    /// both throughput figures — the pipeline's own running time is the
    /// clock they are measured against.
    fn observe(&mut self, session: Session) {
        self.totals = session;
        self.content_rate.observe(
            session.elapsed,
            session.progress.stored_bytes + session.progress.restored_bytes,
        );
        self.transfer_rate
            .observe(session.elapsed, self.transferred);
    }

    fn push_recent(&mut self, line: Activity) {
        self.recent.push_back(line);

        if self.recent.len() > HISTORY_LINES {
            self.recent.pop_front();
        }
    }

    fn summary(&self) -> String {
        let progress = &self.totals.progress;
        let mut parts = Vec::new();

        if progress.scanned > 0 {
            parts.push(format!("{} directories scanned", count(progress.scanned)));
        }

        if progress.stored_files > 0 {
            parts.push(format!(
                "{} files stored ({})",
                count(progress.stored_files),
                human_bytes(progress.stored_bytes)
            ));
        }

        if progress.restored_files > 0 {
            parts.push(format!(
                "{} files restored ({})",
                count(progress.restored_files),
                human_bytes(progress.restored_bytes)
            ));
        }

        if progress.skipped > 0 {
            parts.push(format!("{} skipped", count(progress.skipped)));
        }

        if parts.is_empty() {
            parts.push("no activity".to_string());
        }

        parts.join(", ")
    }

    /// What the run is doing right now, as one line: its state, the time
    /// it has spent working (across every session, not just this one),
    /// and its recent throughput.
    fn status(&self, suspending: bool) -> String {
        // A checkpoint happens *because* the run is suspending, and is
        // the more informative of the two things to say: it can take a
        // while, and it is what makes the interrupted backup restorable
        let state = match (self.phase, suspending) {
            (Phase::Checkpointing, _) => "checkpointing — making what is stored so far restorable",
            (Phase::Finalizing, _) => "finalizing",
            (Phase::Working, true) => "suspending — letting workers finish their current items",
            (Phase::Working, false) => "running",
        };

        let mut parts = vec![state.to_string()];

        if self.totals.elapsed >= Duration::from_secs(1) {
            parts.push(format!("elapsed {}", human_duration(self.totals.elapsed)));
        }

        if let Some(rate) = self.content_rate.per_second() {
            parts.push(format!("content {}/s", human_bytes(rate)));
        }

        if let Some(rate) = self.transfer_rate.per_second() {
            parts.push(format!("repository {}/s", human_bytes(rate)));
        }

        parts.join(" · ")
    }

    /// The skip line: how many were skipped, and the most recent one.
    /// Only the latest is worth screen space — they are reports, not
    /// problems, and a long run produces a great many of them.
    ///
    /// A resumed run counts the skips of earlier sessions without having
    /// a recent one to name, so the two parts are reported separately.
    fn skip_line(&self) -> String {
        let skipped = self.totals.progress.skipped;

        match (&self.last_skip, skipped) {
            (Some(skip), _) => format!("skipped {} · last: {skip}", count(skipped)),
            (None, 0) => "skipped none".to_string(),
            (None, skipped) => format!("skipped {}", count(skipped)),
        }
    }

    /// The stages table flattened onto one line, for the line-oriented
    /// reporter: the same occupancy, backlog, and current item the
    /// dashboard shows in columns.
    fn stage_line(&self) -> Option<String> {
        if self.stages.is_empty() {
            return None;
        }

        let parts: Vec<String> = self
            .stages
            .iter()
            .map(|view| {
                let mut part = match view.threads {
                    Some(threads) => format!(
                        "{} {}/{} busy, {} waiting",
                        view.stage.label(),
                        view.busy.len(),
                        threads,
                        count(view.waiting)
                    ),
                    None => format!("{} {} waiting", view.stage.label(), count(view.waiting)),
                };

                // The two named items, collapsed to one when a stage
                // holds a single item and they are the same
                match (view.newest(), view.oldest()) {
                    (Some(newest), Some(oldest)) if newest.sequence == oldest.sequence => {
                        part.push_str(&format!(" ({})", newest.describe(u16::MAX)));
                    }
                    (Some(newest), Some(oldest)) => {
                        part.push_str(&format!(
                            " ({}, oldest {})",
                            newest.describe(u16::MAX),
                            oldest.describe(u16::MAX)
                        ));
                    }
                    _ => {}
                }

                part
            })
            .collect();

        Some(parts.join(" · "))
    }
}

/// Why a run stopped short of the tree, as the tail of a sentence.
pub fn describe_early_finish(reason: EarlyFinish) -> String {
    match reason {
        EarlyFinish::Requested => "the run was asked to publish what it had and stop".to_string(),
        EarlyFinish::StoreNearlyFull { free, floor } => format!(
            "the store has {} left, {} below the {} this backup was told to \
             leave free",
            human_bytes(free),
            human_bytes(floor.saturating_sub(free)),
            human_bytes(floor)
        ),
        EarlyFinish::StoreAtMaximum { used, limit } => format!(
            "the repository holds {}, at the {} this backup was told to keep \
             it under",
            human_bytes(used),
            human_bytes(limit)
        ),
    }
}

/// Why a limit the run was given is not being applied to it.
pub fn describe_unapplied_limit(limit: StoreLimit) -> String {
    match limit {
        StoreLimit::MinimumFreeSpace => {
            "the store cannot say how much room is left on it, so no room is \
             being left free"
                .to_string()
        }
        StoreLimit::MaximumStoreSize => {
            "the store cannot say how much it is holding, so the repository \
             is not being kept to a size"
                .to_string()
        }
    }
}

fn describe_skip(reason: &SkipReason) -> String {
    match reason {
        SkipReason::Excluded => "excluded".to_string(),
        SkipReason::MountPoint => "mount point".to_string(),
        SkipReason::Vanished => "vanished".to_string(),
        SkipReason::Unreadable(detail) => format!("unreadable: {detail}"),
        SkipReason::SpecialFile => "special file".to_string(),
        SkipReason::UnsafePath => "unsafe path".to_string(),
        SkipReason::Internal => "beeping's own working data".to_string(),
    }
}

pub fn human_bytes(bytes: u64) -> String {
    const UNITS: [&str; 6] = ["B", "KiB", "MiB", "GiB", "TiB", "PiB"];

    let mut value = bytes as f64;
    let mut unit = 0;

    while value >= 1024.0 && unit < UNITS.len() - 1 {
        value /= 1024.0;
        unit += 1;
    }

    if unit == 0 {
        format!("{bytes} B")
    } else {
        format!("{value:.1} {}", UNITS[unit])
    }
}

/// Whole-second durations, since sub-second precision in a figure that
/// counts hours is just noise.
fn human_duration(elapsed: Duration) -> String {
    humantime::format_duration(Duration::from_secs(elapsed.as_secs())).to_string()
}

/// A count with thousands separators: these numbers reach the millions,
/// where an unpunctuated run of digits cannot be read at a glance.
pub fn count(value: u64) -> String {
    let digits = value.to_string();
    let mut out = String::with_capacity(digits.len() + digits.len() / 3);

    for (position, digit) in digits.chars().enumerate() {
        if position > 0 && (digits.len() - position).is_multiple_of(3) {
            out.push(',');
        }

        out.push(digit);
    }

    out
}

/// The newest history lines that fit inside a bordered pane, oldest
/// first, so the pane fills its whole height and appears to scroll.
/// Each is cut to what the pane's width leaves it.
fn visible_tail(lines: &VecDeque<Activity>, area: ratatui::layout::Rect) -> Vec<ListItem<'static>> {
    let height = area.height.saturating_sub(2) as usize; // borders
    let width = area.width.saturating_sub(2);

    lines
        .iter()
        .skip(lines.len().saturating_sub(height))
        .map(|line| ListItem::new(line.render(width)))
        .collect()
}

/// The narrow columns of the stages table, and the single space ratatui
/// puts between each pair of its five columns — what the two path
/// columns have to share the rest of the width with.
const STAGE_COLUMNS: [u16; 3] = [6, 7, 11];
const STAGE_GAPS: u16 = 4;

/// How wide each of the two path columns may be inside a table of this
/// total width, borders included.
fn path_columns(width: u16) -> (u16, u16) {
    let paths = width
        .saturating_sub(2)
        .saturating_sub(STAGE_COLUMNS.iter().sum::<u16>() + STAGE_GAPS);

    (paths / 2, paths - paths / 2)
}

/// Fits a path into a column, keeping its tail.
///
/// Clipping the other end would be worse than useless here: everything
/// on screen is being backed up from the same tree, so the leading
/// components are the part they all share, and the name at the end is
/// the part that says which item this is.
fn fit_path(path: &str, width: u16) -> String {
    let width = width as usize;

    if width == 0 {
        return String::new();
    }

    let length = path.chars().count();

    if length <= width {
        return path.to_string();
    }

    let tail: String = path.chars().skip(length - width + 1).collect();

    format!("…{tail}")
}

/// The stages table: one row per stage, in pipeline order, showing how
/// many of its workers are occupied, how much work is queued for it, and
/// the two items worth naming — the one it took up last and the one it
/// has held longest, which is the one holding it up. The stages without
/// workers of their own name their items too, once finalization reaches
/// them.
fn stages_table(dashboard: &Dashboard, width: u16) -> Table<'_> {
    let dim = Style::default().add_modifier(Modifier::DIM);
    let (newest, oldest) = path_columns(width);

    let rows = dashboard.stages.iter().map(|view| {
        Row::new([
            Cell::from(view.stage.label()),
            Cell::from(Line::from(view.occupancy()).right_aligned()),
            Cell::from(Line::from(count(view.waiting)).right_aligned()),
            Cell::from(
                view.newest()
                    .map_or(String::new(), |item| item.describe(newest)),
            ),
            Cell::from(
                view.oldest()
                    .map_or(String::new(), |item| item.describe(oldest)),
            ),
        ])
    });

    Table::new(
        rows,
        [
            Constraint::Length(STAGE_COLUMNS[0]),
            Constraint::Length(STAGE_COLUMNS[1]),
            Constraint::Length(STAGE_COLUMNS[2]),
            Constraint::Length(newest),
            Constraint::Fill(1),
        ],
    )
    .header(
        Row::new([
            Cell::from("stage"),
            Cell::from(Line::from("busy").right_aligned()),
            Cell::from(Line::from("waiting").right_aligned()),
            Cell::from("newest"),
            Cell::from("oldest"),
        ])
        .style(dim),
    )
    .block(Block::default().borders(Borders::ALL).title("stages"))
}

/// The last `width` samples of a history: as much of it as a pane that
/// wide can draw.
fn window(history: &VecDeque<u64>, width: u16) -> Vec<u64> {
    history
        .iter()
        .skip(history.len().saturating_sub(width as usize))
        .copied()
        .collect()
}

/// A queue's recent depth as one line of block characters, newest at the
/// right, scaled to the tallest sample shown.
///
/// A sample of zero is left blank rather than drawn at the lowest block,
/// so that a queue which ran dry is plainly empty rather than merely
/// nearly so.
fn spark(samples: &[u64], width: u16) -> String {
    const BLOCKS: [char; 8] = ['▁', '▂', '▃', '▄', '▅', '▆', '▇', '█'];

    let peak = samples.iter().copied().max().unwrap_or(0);

    if peak == 0 {
        return String::new();
    }

    // History grows leftward from the newest sample, so a history too
    // short to fill the pane starts part-way across it
    let mut line = " ".repeat((width as usize).saturating_sub(samples.len()));

    for sample in samples {
        line.push(match *sample {
            0 => ' ',
            value => BLOCKS[(value * 7 / peak) as usize],
        });
    }

    line
}

/// The backlog pane: how deep each stage's queue has been over the last
/// few minutes.
///
/// The stages table says how much is waiting now; what it cannot say is
/// which way that number is going, which is the difference between a run
/// that is keeping up and one that is falling behind. Each stage is drawn
/// to its own scale — these queues differ by orders of magnitude — so the
/// peak beside it is what says how deep the tallest column is.
fn backlog_table(dashboard: &Dashboard, width: u16) -> Table<'_> {
    let dim = Style::default().add_modifier(Modifier::DIM);
    let room = width
        .saturating_sub(2)
        .saturating_sub(STAGE_COLUMNS[0] + STAGE_COLUMNS[2] + 2);

    let rows = dashboard.stages.iter().map(move |view| {
        let samples = window(&view.history, room);
        let peak = samples.iter().copied().max().unwrap_or(0);

        Row::new([
            Cell::from(view.stage.label()),
            Cell::from(spark(&samples, room)),
            Cell::from(Line::from(count(peak)).right_aligned()),
        ])
    });

    Table::new(
        rows,
        [
            Constraint::Length(STAGE_COLUMNS[0]),
            Constraint::Fill(1),
            Constraint::Length(STAGE_COLUMNS[2]),
        ],
    )
    .header(
        Row::new([
            Cell::from("stage"),
            Cell::from("backlog"),
            Cell::from(Line::from("peak").right_aligned()),
        ])
        .style(dim),
    )
    .block(
        Block::default()
            .borders(Borders::ALL)
            .title(backlog_span(dashboard, room)),
    )
}

/// The backlog pane's title, which is where its unlabelled horizontal
/// axis gets its scale: the run time between the oldest sample drawn and
/// the newest, which grows as history accumulates and then holds at
/// whatever the pane's width covers.
fn backlog_span(dashboard: &Dashboard, room: u16) -> String {
    let drawn = dashboard
        .stages
        .iter()
        .map(|view| view.history.len())
        .max()
        .unwrap_or(0)
        .min(room as usize);

    // A single column spans nothing yet, and has no scale to give
    match drawn {
        0 | 1 => "backlog".to_string(),
        columns => format!(
            "backlog, last {}",
            human_duration(BACKLOG_SAMPLE * (columns as u32 - 1))
        ),
    }
}

/// What has yet to be visited: the paths each stage will take up next,
/// nearest first. A backup names the directories still to be walked
/// into; a restore names the entries still to be put back.
///
/// Most of a pipeline's backlog is deliberately absent. Files the scan
/// has already found, chunks in flight to the backend, and the record
/// of what is already stored are all work the run has in hand — naming
/// them would bury the frontier under thousands of lines that say
/// nothing about where the run is going. An empty pane means the stages
/// that do name their queues have nothing waiting.
///
/// The stages come in pipeline order, and a pane too short for all of
/// them lets them take turns claiming a line, so that a stage with a
/// short queue is not buried under one with a long one. The last line
/// accounts for everything waiting behind what is named — the whole
/// depth of those queues, not merely the part the pipeline sends by
/// name.
fn pending_lines(dashboard: &Dashboard, area: ratatui::layout::Rect) -> Vec<ListItem<'static>> {
    let height = area.height.saturating_sub(2) as usize; // borders
    let width = area.width.saturating_sub(2);

    if height == 0 {
        return Vec::new();
    }

    let queues: Vec<(&'static str, &[String])> = dashboard
        .stages
        .iter()
        .filter_map(|view| {
            view.upcoming
                .as_deref()
                .map(|paths| (view.stage.label(), paths))
        })
        .collect();

    let waiting: u64 = dashboard
        .stages
        .iter()
        .filter(|view| view.upcoming.is_some())
        .map(|view| view.waiting)
        .sum();

    let named: usize = queues.iter().map(|(_, paths)| paths.len()).sum();

    // Room for every name, or for all but the last line, which says how
    // much is waiting behind them — except in a pane one line tall,
    // where naming one path is worth more than counting the rest
    let room = if waiting as usize > height && height > 1 {
        named.min(height - 1)
    } else {
        named.min(height)
    };

    let mut share = vec![0usize; queues.len()];
    let mut given = 0;

    while given < room {
        for (queue, (_, paths)) in queues.iter().enumerate() {
            if given == room {
                break;
            }

            if share[queue] < paths.len() {
                share[queue] += 1;
                given += 1;
            }
        }
    }

    let mut lines: Vec<ListItem<'static>> = Vec::with_capacity(height);

    for (queue, (label, paths)) in queues.iter().enumerate() {
        for path in &paths[..share[queue]] {
            lines.push(ListItem::new(format!(
                "{label:<column$} {}",
                fit_path(path, width.saturating_sub(STAGE_COLUMNS[0] + 1)),
                column = STAGE_COLUMNS[0] as usize
            )));
        }
    }

    if waiting > given as u64 && lines.len() < height {
        lines.push(ListItem::new(format!(
            "… and {} more",
            count(waiting - given as u64)
        )));
    }

    lines
}

/// Runs the full-screen dashboard until the pipeline finishes, returning
/// a final summary line. Pressing `s`, `q`, Esc, or Ctrl-C requests
/// suspension.
pub fn dashboard(
    title: &str,
    events: Receiver<Event>,
    suspend: Arc<AtomicBool>,
) -> std::io::Result<String> {
    let mut terminal = ratatui::init();
    let result = dashboard_loop(title, &events, &suspend, &mut terminal);
    ratatui::restore();

    result
}

fn dashboard_loop(
    title: &str,
    events: &Receiver<Event>,
    suspend: &AtomicBool,
    terminal: &mut ratatui::DefaultTerminal,
) -> std::io::Result<String> {
    let mut dashboard = Dashboard::default();
    let mut finished = false;

    while !finished {
        loop {
            match events.try_recv() {
                Ok(event) => dashboard.apply(event),
                Err(crossbeam_channel::TryRecvError::Empty) => break,
                Err(crossbeam_channel::TryRecvError::Disconnected) => {
                    finished = true;
                    break;
                }
            }
        }

        let suspending = suspend.load(Ordering::Acquire);

        terminal.draw(|frame| draw(frame, title, &dashboard, suspending))?;

        if crossterm::event::poll(Duration::from_millis(100))?
            && let TermEvent::Key(key) = crossterm::event::read()?
        {
            let ctrl_c =
                key.code == KeyCode::Char('c') && key.modifiers.contains(KeyModifiers::CONTROL);

            if matches!(key.code, KeyCode::Char('s' | 'q') | KeyCode::Esc) || ctrl_c {
                suspend.store(true, Ordering::Release);
            }
        }
    }

    Ok(dashboard.summary())
}

/// Draws one frame: the run's totals, its stages, and three panes — what
/// has been finished, how the queues ahead of it are trending, and what
/// its workers have in their hands.
fn draw(frame: &mut Frame, title: &str, dashboard: &Dashboard, suspending: bool) {
    // Both stage tables are exactly as tall as they need to be
    // — borders, a header, and one row per reported stage — so
    // the activity log and the pending pane get what is left.
    let stages_height = match dashboard.stages.len() {
        0 => 0,
        rows => rows as u16 + 3,
    };

    let [header, counters, stages, panes, skipped, footer] = Layout::vertical([
        Constraint::Length(1),
        Constraint::Length(4),
        Constraint::Length(stages_height),
        Constraint::Fill(1),
        Constraint::Length(1),
        Constraint::Length(1),
    ])
    .areas(frame.area());

    // What has happened on the left; what is happening and what
    // is coming on the right
    let [recent, current] =
        Layout::horizontal([Constraint::Fill(1), Constraint::Fill(1)]).areas(panes);
    // The backlog pane wants to be as tall as the stages table, and is
    // what gives way when the window is too short to leave the pending
    // pane anything worth showing
    let [backlog, pending] =
        Layout::vertical([Constraint::Max(stages_height), Constraint::Min(5)]).areas(current);

    let dim = Style::default().add_modifier(Modifier::DIM);

    frame.render_widget(
        Paragraph::new(Line::from(title.to_string()))
            .style(Style::default().add_modifier(Modifier::BOLD)),
        header,
    );

    frame.render_widget(
        Paragraph::new(format!(
            "{}\nstatus: {}",
            dashboard.summary(),
            dashboard.status(suspending)
        ))
        .block(Block::default().borders(Borders::ALL).title("progress")),
        counters,
    );

    frame.render_widget(stages_table(dashboard, stages.width), stages);

    frame.render_widget(
        List::new(visible_tail(&dashboard.recent, recent))
            .block(Block::default().borders(Borders::ALL).title("activity")),
        recent,
    );

    frame.render_widget(backlog_table(dashboard, backlog.width), backlog);

    frame.render_widget(
        List::new(pending_lines(dashboard, pending))
            .block(Block::default().borders(Borders::ALL).title("pending")),
        pending,
    );

    frame.render_widget(Paragraph::new(dashboard.skip_line()).style(dim), skipped);

    frame.render_widget(
        Paragraph::new("s / q / Esc / Ctrl-C: suspend (resume with the same command)"),
        footer,
    );
}

/// Line-oriented reporter: skips are printed as they happen, progress
/// every couple of seconds, and the final summary at the end.
pub fn plain(events: Receiver<Event>) -> String {
    let mut dashboard = Dashboard::default();
    let mut last_progress = Instant::now();
    let stderr = std::io::stderr();

    for event in events {
        if let Event::Skipped { path, reason } = &event {
            let _ = writeln!(
                stderr.lock(),
                "skipped {} ({})",
                path.display(),
                describe_skip(reason)
            );
        }

        match event {
            Event::Checkpointing => {
                let _ = writeln!(
                    stderr.lock(),
                    "checkpointing: making what is stored so far restorable..."
                );
            }
            Event::Finalizing => {
                let _ = writeln!(stderr.lock(), "finalizing...");
            }
            Event::FinishingEarly(reason) => {
                let _ = writeln!(
                    stderr.lock(),
                    "finishing early: {}",
                    describe_early_finish(reason)
                );
            }
            Event::LimitNotApplied(limit) => {
                let _ = writeln!(
                    stderr.lock(),
                    "warning: {}",
                    describe_unapplied_limit(limit)
                );
            }
            _ => {}
        }

        dashboard.apply(event);

        if last_progress.elapsed() >= Duration::from_secs(2) {
            let mut line = format!("{} · {}\n", dashboard.summary(), dashboard.status(false));

            // The stages, indented under the totals they belong to
            if let Some(stages) = dashboard.stage_line() {
                line.push_str("  ");
                line.push_str(&stages);
                line.push('\n');
            }

            let _ = write!(stderr.lock(), "{line}");
            last_progress = Instant::now();
        }
    }

    dashboard.summary()
}

#[cfg(test)]
mod tests {
    use super::*;
    use beeping::pipeline::{Progress, StageBacklog};
    use std::path::PathBuf;

    fn lines(count: usize) -> VecDeque<Activity> {
        (0..count)
            .map(|n| Activity::Note(format!("line-{n}")))
            .collect()
    }

    fn texts(items: &[ListItem<'_>]) -> Vec<String> {
        // ListItem has no accessor; rebuild via Text conversion
        items.iter().map(|item| format!("{:?}", item)).collect()
    }

    /// The paths the two columns would name, for comparing at a glance.
    fn named(view: &StageView) -> (Option<&str>, Option<&str>) {
        (
            view.newest().map(|item| item.path.as_str()),
            view.oldest().map(|item| item.path.as_str()),
        )
    }

    /// A stage's backlog with nothing named: a queue of something other
    /// than paths waiting for a worker.
    fn backlog(stage: Stage, waiting: u64) -> StageBacklog {
        StageBacklog {
            stage,
            waiting,
            upcoming: None,
        }
    }

    /// A stage's backlog with the head of its queue named.
    fn queued(stage: Stage, waiting: u64, upcoming: &[&str]) -> StageBacklog {
        StageBacklog {
            stage,
            waiting,
            upcoming: Some(upcoming.iter().map(PathBuf::from).collect()),
        }
    }

    fn began(stage: Stage, worker: usize, path: &str) -> Event {
        Event::Began {
            stage,
            worker,
            path: PathBuf::from(path),
        }
    }

    #[test]
    fn tall_panes_fill_with_the_newest_lines() {
        let area = ratatui::layout::Rect::new(0, 0, 80, 22); // 20 inner rows
        let history = lines(50);
        let items = visible_tail(&history, area);

        assert_eq!(items.len(), 20, "the pane's full height is used");

        let rendered = texts(&items);
        assert!(
            rendered.first().unwrap().contains("line-30"),
            "oldest visible"
        );
        assert!(rendered.last().unwrap().contains("line-49"), "newest last");
    }

    #[test]
    fn short_histories_show_everything() {
        let area = ratatui::layout::Rect::new(0, 0, 80, 22);
        let history = lines(5);
        let items = visible_tail(&history, area);

        assert_eq!(items.len(), 5);
    }

    #[test]
    fn degenerate_panes_render_nothing() {
        let history = lines(10);

        for height in [0, 1, 2] {
            let area = ratatui::layout::Rect::new(0, 0, 80, height);
            assert!(visible_tail(&history, area).len() <= height as usize);
        }
    }

    /// The activity log's own pane is half the width it used to have, so
    /// its lines are fitted like the table's are: the verb whole, and as
    /// much of the path's tail as the rest of the line holds.
    #[test]
    fn activity_lines_keep_the_verb_and_the_end_of_the_path() {
        let stored = Activity::Item {
            verb: "stored",
            path: "/home/djarb/models/big.safetensors".to_string(),
        };

        assert_eq!(
            stored.render(60),
            "stored   /home/djarb/models/big.safetensors"
        );
        assert_eq!(stored.render(24), "stored   …ig.safetensors");

        // Prose is not a path, and front-truncating it would only hide
        // the part that says what it is about
        let note = Activity::Note("resumed: totals include previous sessions".to_string());
        assert_eq!(note.render(20), "resumed: totals include previous sessions");
    }

    /// A run that stops short says why in the log, where it stays put:
    /// nothing else is happening by then, and finalizing a large
    /// manifest can take long enough for the question to come up.
    #[test]
    fn an_early_finish_says_why_in_the_activity_log() {
        let mut dashboard = Dashboard::default();

        dashboard.apply(Event::FinishingEarly(EarlyFinish::StoreNearlyFull {
            free: 3 << 30,
            floor: 20 << 30,
        }));
        dashboard.apply(Event::Finalizing);

        let note = dashboard.recent.back().unwrap().render(80);

        assert_eq!(
            note,
            "finishing early: the store has 3.0 GiB left, 17.0 GiB below the \
             20.0 GiB this backup was told to leave free"
        );

        // and the run carries on to publish what it has
        assert!(dashboard.status(false).starts_with("finalizing"));

        // The other limit on the store reads from the other end
        let mut dashboard = Dashboard::default();

        dashboard.apply(Event::FinishingEarly(EarlyFinish::StoreAtMaximum {
            used: 500 << 30,
            limit: 500 << 30,
        }));

        assert_eq!(
            dashboard.recent.back().unwrap().render(80),
            "finishing early: the repository holds 500.0 GiB, at the 500.0 \
             GiB this backup was told to keep it under"
        );
    }

    /// A limit that cannot be applied is worth a line of its own: the
    /// run looks exactly like one being held to it otherwise.
    #[test]
    fn a_limit_that_cannot_be_applied_says_so() {
        let mut dashboard = Dashboard::default();

        dashboard.apply(Event::LimitNotApplied(StoreLimit::MaximumStoreSize));

        assert_eq!(
            dashboard.recent.back().unwrap().render(80),
            "the store cannot say how much it is holding, so the repository \
             is not being kept to a size"
        );
    }

    #[test]
    fn counts_are_grouped_in_threes() {
        assert_eq!(count(0), "0");
        assert_eq!(count(999), "999");
        assert_eq!(count(1000), "1,000");
        assert_eq!(count(12345), "12,345");
        assert_eq!(count(1234567890), "1,234,567,890");
    }

    #[test]
    fn stages_appear_in_the_order_the_pipeline_mentions_them() {
        let mut dashboard = Dashboard::default();

        for (stage, threads) in [(Stage::Scan, 2), (Stage::Chunk, 4), (Stage::Upload, 3)] {
            dashboard.apply(Event::Staffing { stage, threads });
        }

        dashboard.apply(Event::Backlog(vec![backlog(Stage::Record, 17)]));

        let order: Vec<Stage> = dashboard.stages.iter().map(|view| view.stage).collect();
        assert_eq!(
            order,
            [Stage::Scan, Stage::Chunk, Stage::Upload, Stage::Record]
        );

        // A stage nobody staffs is worked by finalization
        assert_eq!(dashboard.stages[3].occupancy(), "—");
        assert_eq!(dashboard.stages[3].waiting, 17);
    }

    #[test]
    fn occupancy_follows_workers_in_and_out_of_work() {
        let mut dashboard = Dashboard::default();

        dashboard.apply(Event::Staffing {
            stage: Stage::Chunk,
            threads: 3,
        });
        dashboard.apply(began(Stage::Chunk, 0, "a.dat"));
        dashboard.apply(began(Stage::Chunk, 1, "b.dat"));

        assert_eq!(dashboard.stages[0].occupancy(), "2/3");

        // Taking the next item replaces the worker's previous one
        dashboard.apply(began(Stage::Chunk, 1, "c.dat"));
        assert_eq!(dashboard.stages[0].occupancy(), "2/3");

        dashboard.apply(Event::Idle {
            stage: Stage::Chunk,
            worker: 0,
        });
        assert_eq!(dashboard.stages[0].occupancy(), "1/3");

        dashboard.apply(Event::Idle {
            stage: Stage::Chunk,
            worker: 1,
        });
        assert_eq!(dashboard.stages[0].occupancy(), "0/3");
        assert_eq!(named(&dashboard.stages[0]), (None, None));
    }

    #[test]
    fn a_stage_names_both_where_it_is_and_what_holds_it_up() {
        let mut dashboard = Dashboard::default();

        dashboard.apply(began(Stage::Chunk, 0, "slow.iso"));
        dashboard.apply(began(Stage::Chunk, 1, "quick-a.txt"));
        dashboard.apply(began(Stage::Chunk, 1, "quick-b.txt"));

        assert_eq!(
            named(&dashboard.stages[0]),
            (Some("quick-b.txt"), Some("slow.iso")),
            "where it has got to, and what is holding it up"
        );

        // A finished item is named by neither: with the fast worker
        // idle, one item is in hand and both columns name it
        dashboard.apply(Event::Idle {
            stage: Stage::Chunk,
            worker: 1,
        });
        assert_eq!(
            named(&dashboard.stages[0]),
            (Some("slow.iso"), Some("slow.iso"))
        );

        dashboard.apply(Event::Idle {
            stage: Stage::Chunk,
            worker: 0,
        });
        assert_eq!(
            named(&dashboard.stages[0]),
            (None, None),
            "an idle stage names nothing rather than a stale item"
        );
    }

    #[test]
    fn progress_belongs_to_the_item_in_hand() {
        let mut dashboard = Dashboard::default();

        dashboard.apply(began(Stage::Chunk, 0, "big.iso"));
        dashboard.apply(Event::Progressed {
            stage: Stage::Chunk,
            worker: 0,
            done: 43,
            total: 100,
        });

        let item = dashboard.stages[0].oldest().unwrap();
        assert_eq!(item.describe(40), " 43% big.iso");

        // The next item starts afresh rather than inheriting a
        // percentage from the last one
        dashboard.apply(began(Stage::Chunk, 0, "next.iso"));
        assert_eq!(
            dashboard.stages[0].oldest().unwrap().describe(40),
            "next.iso"
        );

        // A percentage keeps its width; only the path gives way
        dashboard.apply(Event::Progressed {
            stage: Stage::Chunk,
            worker: 0,
            done: 7,
            total: 1000,
        });
        let cell = dashboard.stages[0].oldest().unwrap().describe(10);
        assert_eq!(cell, "  0% ….iso");
        assert_eq!(cell.chars().count(), 10, "the column is not overrun");

        // Progress for a worker holding nothing is simply dropped
        dashboard.apply(Event::Idle {
            stage: Stage::Chunk,
            worker: 0,
        });
        dashboard.apply(Event::Progressed {
            stage: Stage::Chunk,
            worker: 0,
            done: 1,
            total: 2,
        });
        assert_eq!(named(&dashboard.stages[0]), (None, None));
    }

    #[test]
    fn paths_are_cut_at_the_front_where_they_are_all_alike() {
        assert_eq!(fit_path("sub/file.dat", 20), "sub/file.dat");
        assert_eq!(fit_path("sub/file.dat", 12), "sub/file.dat");
        assert_eq!(fit_path("sub/file.dat", 8), "…ile.dat");
        assert_eq!(fit_path("sub/file.dat", 1), "…");
        assert_eq!(fit_path("sub/file.dat", 0), "");

        // Cut on char boundaries, not byte offsets
        let accented = "café/ünïcøde.dat";
        assert_eq!(accented.chars().count(), 16);
        assert_eq!(fit_path(accented, 8), "…øde.dat");
        assert_eq!(fit_path(accented, 8).chars().count(), 8);
    }

    #[test]
    fn the_two_path_columns_share_what_the_narrow_ones_leave() {
        let (newest, oldest) = path_columns(110);
        assert_eq!(newest + oldest, 110 - 2 - 24 - 4);
        assert!(oldest >= newest, "odd remainders go to the last column");

        // A cramped table asks for no path width rather than underflowing
        assert_eq!(path_columns(20), (0, 0));
    }

    #[test]
    fn only_the_most_recent_skip_is_kept() {
        let mut dashboard = Dashboard::default();

        for name in ["one", "two", "three"] {
            dashboard.apply(Event::Skipped {
                path: PathBuf::from(name),
                reason: SkipReason::Excluded,
            });
        }

        dashboard.apply(Event::Totals(Session {
            progress: Progress {
                skipped: 3,
                ..Progress::default()
            },
            elapsed: Duration::from_secs(1),
        }));

        assert_eq!(dashboard.skip_line(), "skipped 3 · last: three (excluded)");

        // A resumed run inherits the count without inheriting a path
        let mut resumed = Dashboard::default();
        assert_eq!(resumed.skip_line(), "skipped none");

        resumed.apply(Event::Resumed(Session {
            progress: Progress {
                skipped: 1200,
                ..Progress::default()
            },
            elapsed: Duration::from_secs(30),
        }));
        assert_eq!(resumed.skip_line(), "skipped 1,200");
        assert!(
            dashboard.recent.is_empty(),
            "skips do not crowd the activity log"
        );
    }

    #[test]
    fn the_line_reporter_says_what_every_stage_is_doing() {
        let mut dashboard = Dashboard::default();

        assert_eq!(dashboard.stage_line(), None, "before the pipeline speaks");

        dashboard.apply(Event::Staffing {
            stage: Stage::Chunk,
            threads: 16,
        });
        dashboard.apply(began(Stage::Chunk, 4, "big.iso"));
        dashboard.apply(Event::Backlog(vec![
            backlog(Stage::Chunk, 0),
            backlog(Stage::Record, 1489),
        ]));

        assert_eq!(
            dashboard.stage_line().unwrap(),
            "chunk 1/16 busy, 0 waiting (big.iso) · record 1,489 waiting"
        );
    }

    /// Each queue is drawn against its own peak, since a scan backlog of
    /// half a million and an upload backlog of eight share a pane.
    #[test]
    fn a_backlog_is_drawn_against_its_own_peak() {
        assert_eq!(spark(&[0, 1, 5, 10], 4), " ▁▄█");
        assert_eq!(spark(&[0, 100, 500, 1000], 4), " ▁▄█");

        // A queue that has been empty throughout has nothing to say
        assert_eq!(spark(&[0, 0, 0], 8), "");

        // History grows leftward from the newest sample
        assert_eq!(spark(&[2, 1], 6), "    █▄");

        // A pane narrower than the history shows the recent end of it,
        // which is what the caller's window hands over
        let history: VecDeque<u64> = (1..=10).collect();
        assert_eq!(window(&history, 3), vec![8, 9, 10]);
        assert_eq!(window(&history, 40), (1..=10).collect::<Vec<u64>>());
    }

    /// The display is fed several times a second; the backlog history is
    /// not, or the pane would hold a few seconds of noise.
    #[test]
    fn backlog_history_is_sampled_on_the_pipelines_clock() {
        let mut dashboard = Dashboard::default();

        let report = |dashboard: &mut Dashboard, millis: u64, waiting: u64| {
            dashboard.apply(Event::Totals(Session {
                progress: Progress::default(),
                elapsed: Duration::from_millis(millis),
            }));
            dashboard.apply(Event::Backlog(vec![backlog(Stage::Chunk, waiting)]));
        };

        for tick in 0..=50 {
            report(&mut dashboard, tick * 200, tick);
        }

        let view = &dashboard.stages[0];

        // Ten seconds of reports, two seconds to the sample
        assert_eq!(view.history.len(), 6);
        assert_eq!(view.waiting, 50, "the depth itself is always current");
        assert_eq!(*view.history.back().unwrap(), 50);

        // A second pipeline reporting into the same display starts the
        // history over rather than continuing the first one's
        report(&mut dashboard, 0, 7);
        let view = &dashboard.stages[0];

        assert_eq!(view.history.len(), 1);
        assert_eq!(*view.history.back().unwrap(), 7);
    }

    #[test]
    fn transfers_and_finished_content_are_measured_separately() {
        let mut dashboard = Dashboard::default();

        let totals = |elapsed, stored_bytes| {
            Event::Totals(Session {
                progress: Progress {
                    stored_bytes,
                    ..Progress::default()
                },
                elapsed: Duration::from_secs(elapsed),
            })
        };

        // One huge file: chunks keep flowing, but nothing finishes
        dashboard.apply(totals(0, 0));

        for second in 1..=4 {
            dashboard.apply(Event::Transferred { bytes: 3000 });
            dashboard.apply(totals(second, 0));
        }

        assert_eq!(dashboard.content_rate.per_second(), Some(0));
        assert_eq!(dashboard.transfer_rate.per_second(), Some(3000));

        let status = dashboard.status(false);
        assert!(status.contains("repository 2.9 KiB/s"), "{status}");
    }

    #[test]
    fn throughput_is_measured_over_the_recent_window() {
        let mut rate = Rate::default();

        assert_eq!(rate.per_second(), None, "nothing sampled yet");

        rate.observe(Duration::from_secs(0), 0);
        assert_eq!(rate.per_second(), None, "one sample is not a rate");

        for second in 1..=4 {
            rate.observe(Duration::from_secs(second), second * 1000);
        }

        assert_eq!(rate.per_second(), Some(1000));

        // Ten seconds of double speed pushes the old samples out of the
        // window, leaving the current rate rather than the average
        for second in 5..=14 {
            rate.observe(Duration::from_secs(second), 4000 + (second - 4) * 2000);
        }

        assert_eq!(rate.per_second(), Some(2000));
    }

    #[test]
    fn a_stalled_run_reports_no_throughput_rather_than_a_stale_one() {
        let mut rate = Rate::default();

        for second in 0..=3 {
            rate.observe(Duration::from_secs(second), second * 1000);
        }

        // The bytes stop moving; the window fills with flat samples
        for second in 4..=20 {
            rate.observe(Duration::from_secs(second), 3000);
        }

        assert_eq!(rate.per_second(), Some(0));
    }

    /// The pending pane names what each queue will hand over next, and
    /// accounts for everything waiting behind the names it has room for.
    ///
    /// The dashboard serves both pipelines and knows neither, so what it
    /// draws depends only on which stages name their queues — here, a
    /// restore's two, beside a stage whose backlog is not a queue of
    /// paths at all.
    #[test]
    fn the_pending_pane_names_what_the_queues_will_hand_out_next() {
        let mut dashboard = Dashboard::default();

        assert!(
            pending_lines(&dashboard, ratatui::layout::Rect::new(0, 0, 40, 8)).is_empty(),
            "nothing queued yet"
        );

        dashboard.apply(Event::Backlog(vec![
            queued(Stage::Write, 2, &["/tree/src", "/tree/doc"]),
            queued(
                Stage::Apply,
                400,
                &["/tree/a.dat", "/tree/b.dat", "/tree/c.dat"],
            ),
            backlog(Stage::Record, 9000),
        ]));

        let pane = ratatui::layout::Rect::new(0, 0, 40, 8); // 6 inner rows
        let rendered = texts(&pending_lines(&dashboard, pane));

        assert_eq!(rendered.len(), 6);
        assert!(rendered[0].contains("write"), "{:?}", rendered[0]);
        assert!(rendered[0].contains("src"), "nearest first");
        assert!(rendered[1].contains("doc"));
        assert!(rendered[2].contains("apply"), "{:?}", rendered[2]);
        assert!(rendered[2].contains("a.dat"), "then the next stage along");
        assert!(
            rendered[5].contains("and 397 more"),
            "the rest of the two queues — and not the record queue, which \
             holds what is already stored: {:?}",
            rendered[5]
        );

        // Too short for all the names: every queue still gets a turn,
        // and the last line grows to cover what is left out
        let short = ratatui::layout::Rect::new(0, 0, 40, 4); // 2 inner rows
        let rendered = texts(&pending_lines(&dashboard, short));

        assert_eq!(rendered.len(), 2);
        assert!(rendered[0].contains("src"), "{:?}", rendered[0]);
        assert!(rendered[1].contains("and 401 more"), "{:?}", rendered[1]);

        // One line is better spent naming something than counting
        let sliver = ratatui::layout::Rect::new(0, 0, 40, 3); // 1 inner row
        let rendered = texts(&pending_lines(&dashboard, sliver));

        assert_eq!(rendered.len(), 1);
        assert!(rendered[0].contains("src"), "{:?}", rendered[0]);

        // A queue that drains takes its names off the pane with it
        dashboard.apply(Event::Backlog(vec![
            queued(Stage::Write, 0, &[]),
            queued(Stage::Apply, 0, &[]),
        ]));

        assert!(pending_lines(&dashboard, pane).is_empty());
    }

    /// The whole frame, as characters, for tests that care about how the
    /// panes divide the screen rather than about one pane's contents.
    fn rendered(dashboard: &Dashboard, width: u16, height: u16) -> Vec<String> {
        let mut terminal =
            ratatui::Terminal::new(ratatui::backend::TestBackend::new(width, height)).unwrap();

        terminal
            .draw(|frame| draw(frame, "beeping backup", dashboard, false))
            .unwrap();

        let buffer = terminal.backend().buffer().clone();

        (0..height)
            .map(|y| {
                (0..width)
                    .map(|x| buffer[(x, y)].symbol())
                    .collect::<String>()
            })
            .collect()
    }

    /// The panes below the stages table divide the screen in two: what
    /// has been finished on the left, and what the queues have been
    /// doing and will do next on the right.
    #[test]
    fn the_lower_panes_split_the_width_between_history_and_what_is_ahead() {
        let mut dashboard = Dashboard::default();

        dashboard.apply(Event::Staffing {
            stage: Stage::Scan,
            threads: 2,
        });
        dashboard.apply(began(Stage::Scan, 0, "/tree/models"));
        dashboard.apply(Event::Stored {
            path: PathBuf::from("/tree/notes/a.txt"),
            bytes: 12,
        });

        for tick in 0..=10 {
            dashboard.apply(Event::Totals(Session {
                progress: Progress {
                    scanned: 40 + tick,
                    stored_files: 300 + tick * 9,
                    stored_bytes: (12 + tick) * 1024 * 1024,
                    ..Progress::default()
                },
                elapsed: Duration::from_secs(tick * 2),
            }));
            dashboard.apply(Event::Backlog(vec![queued(
                Stage::Scan,
                tick * 40,
                &["/tree/notes", "/tree/models"],
            )]));
        }

        let screen = rendered(&dashboard, 100, 20);

        // A pane's title is where its top border opens, which is also
        // the only place these words appear as titles
        let titled = |name: &str| {
            let opens = format!("┌{name}");

            screen
                .iter()
                .position(|line| line.contains(&opens))
                .unwrap_or_else(|| panic!("no {name} pane in\n{}", screen.join("\n")))
        };

        // activity and backlog share a row of titles; pending is under
        // the backlog pane, which is as tall as the stages table
        assert_eq!(titled("activity"), titled("backlog"));
        assert!(titled("pending") > titled("backlog"));

        let titles = &screen[titled("activity")];
        assert!(
            titles.contains("backlog, last 20s"),
            "the pane says how much of the run it is drawing: {titles}"
        );

        let split = titles.split("┌backlog").next().unwrap().chars().count();
        assert!((45..55).contains(&split), "{titles}");

        // Each pane holds what belongs to it, and nothing has spilled
        assert!(screen.iter().any(|line| line.contains("stored")));
        assert!(
            screen[titled("pending") + 1].contains("/tree/notes"),
            "{}",
            screen[titled("pending") + 1]
        );
        // Title, header, then the one stage this run has reported
        assert!(
            screen[titled("backlog") + 2].contains('█'),
            "a queue that has been growing all run should show it: {}",
            screen[titled("backlog") + 2]
        );
    }

    #[test]
    fn totals_come_from_the_pipeline_rather_than_being_re_derived() {
        let mut dashboard = Dashboard::default();

        // Per-item events feed the activity log, not the counters
        dashboard.apply(Event::Stored {
            path: PathBuf::from("a.dat"),
            bytes: 10,
        });
        assert_eq!(dashboard.summary(), "no activity");

        dashboard.apply(Event::Totals(Session {
            progress: Progress {
                scanned: 12,
                stored_files: 3400,
                stored_bytes: 5 * 1024 * 1024,
                ..Progress::default()
            },
            elapsed: Duration::from_secs(90),
        }));

        assert_eq!(
            dashboard.summary(),
            "12 directories scanned, 3,400 files stored (5.0 MiB)"
        );
        assert_eq!(
            dashboard.status(false),
            "running · elapsed 1m 30s",
            "one sample in is too early for a throughput figure"
        );
        assert!(dashboard.status(true).starts_with("suspending"));
    }
}