pinch-points 0.4.0

//! Online versus, shell side: a UDP lockstep session driving the same
//! versus mode local play uses (spec §7.6 fallback path; see sim::net for
//! why lockstep rather than GGRS rollback today).
//!
//! Dev-grade session setup: `PINCH_HOST=<port>` hosts as player 0,
//! `PINCH_JOIN=<addr:port>` joins as player 1. Both boot straight into the
//! arena; the match starts once the handshake completes.

use crate::sim::{HASH_INTERVAL, Lockstep, MAX_PLAYERS, PlayerAction};
use crate::transport::{
    Announcer, MatchTerms, NetMsg, SeriesStanding, UdpTransport, name_from_wire, table_from_wire,
    wire_table,
};
use bevy::prelude::*;
pub use peers::{Peer, PeerBook, Place};
use std::collections::VecDeque;

/// How many ticks a resume is repeated after the local session unpauses.
/// A second's worth: enough to ride out a loss burst, and harmless if the
/// peers already heard the first one.
const RESUME_ECHOES: u8 = 30;

pub struct OnlineSession {
    pub transport: UdpTransport,
    pub session: Lockstep,
    /// Seats in the match, humans plus AI. The lockstep only carries the
    /// humans, so this is not `session.player_count()`.
    pub seats: u8,
    /// What the lobby agreed the match is: AI seats, map, gulls, round
    /// length, light, team scoring, and the board's seed. Every peer builds
    /// from these, so nothing about the round is a local opinion.
    pub terms: MatchTerms,
    /// The host's handmade beach, compressed, when the round is played on
    /// one. A generated arena travels as a seed in [`Self::terms`]; a
    /// level somebody built has to travel as itself, because no peer but
    /// the host has the file. Empty for the built-in maps.
    pub beach: Vec<u8>,
    /// The peers on the socket, one row each: the chair the launch plan
    /// gave it (`Queued` for one that turned up mid-round, and for every
    /// peer a joiner knows, since a joiner keeps no plan and answers
    /// nobody), what it calls itself, whether it asked to watch, and how
    /// long since it last said anything.
    ///
    /// The silence is the difference between a player who has *gone* and
    /// one who is merely behind: a machine that is still talking to us has
    /// not left the room, however far its inputs have fallen back, and
    /// giving its castle away because a burst of loss held one frame up is
    /// how you lose a friend's round for them. A joiner keeps the one row
    /// it has, the host, and calls the round off when that goes quiet.
    pub(crate) peers: PeerBook,
    /// What each seat is called, agreed at the handshake so every peer
    /// shows the same table. Empty entries fall back to seat labels; local
    /// couch names never apply to an online round.
    pub names: [String; MAX_PLAYERS],
    /// The peers' state hashes against our own, and the first frame they
    /// disagreed on, if ever.
    hashes: HashCheck,
    /// Whether the picture has stopped moving, and on whom.
    stall: StallWatch,
    /// Seats given up on this round, each with the frame it was emptied
    /// from, so the shell can put an AI in each and say so once rather
    /// than every frame, and the host can keep telling the table.
    pub abandoned: Vec<(u8, u32)>,
    /// The beach this session came from, and what it owes it.
    pub home: Home,
    /// A next round has been agreed and the session is armed for it; the
    /// shell reads this and walks everyone back into the arena. Cleared by
    /// whoever acts on it.
    pub next_round: bool,
    /// Where the series stands as the next round begins, as the host said
    /// it: the 1-based round number and the wins per seat, both re-dealt to
    /// this round's chairs. The shell folds it into its `Tournament` on the
    /// way into the arena, so a peer admitted mid-series joins the table's
    /// standings rather than starting its own, and a survivor whose seat
    /// moved keeps the wins it earned. `None` outside a series.
    pub series_standing: Option<SeriesStanding>,
    /// Ticks of `Resume` still to repeat (see [`RESUME_ECHOES`]).
    resume_echo: u8,
}

/// The beach a session came from: its beacon, what it is called, and
/// whether there is a lobby to walk the table back to.
pub struct Home {
    /// Host side: the beacon, carried out of the lobby so the beach stays
    /// on the air while the round runs, listed as in progress with its
    /// occupancy, for anyone who wants the next one. Silent on a joiner,
    /// and on a direct `PINCH_HOST` pair, which never announced at all.
    announcer: Farewell,
    /// What the beach is called, for the beacon it keeps up while the round
    /// runs. Not a player's name: the list is choosing between games.
    pub game_name: String,
    announce_in: f32,
    /// Whether this session was formed in the beach lobby, which is the
    /// only place a finished match has to walk its table back to: the
    /// direct `PINCH_HOST` pair was never anywhere else to begin with.
    pub from_lobby: bool,
    /// Joiner side: seconds until the next greeting while the results card
    /// is up.
    ///
    /// An established session says nothing at all between rounds. Inputs
    /// and hashes belong to the round that ended, and the host has nothing
    /// to send until somebody calls the next one, so both ends fall silent
    /// on purpose. That is fine until silence is being read as evidence,
    /// and then it is a joiner walking out of a perfectly good table
    /// twenty seconds into a results card. So a joiner keeps saying hello,
    /// on the same cadence the lobby uses, and the host's reply is what
    /// tells it the host is still there.
    greet_in: f32,
}

impl Home {
    /// Nowhere in particular: the direct pair, which never announced and
    /// has no lobby behind it.
    fn nowhere() -> Home {
        Home {
            announcer: Farewell::silent(),
            game_name: String::new(),
            announce_in: 0.0,
            from_lobby: false,
            greet_in: 0.0,
        }
    }
}

/// Whether the round has stopped moving, for how long, and who the status
/// line is naming for it.
///
/// Measured by the frame number rather than by whether the next frame's
/// inputs happen to be in, because the second question is asked at a
/// moment where the answer is always no. The sim runs on the fixed step
/// and advances until it *cannot*; the tick that watches for stalls runs
/// after it, in `Update`. So it always found the next frame's slots
/// empty and always called a healthy round stalled. That put "waiting
/// for Bob" on both screens of a round that was running perfectly, and
/// made the socket-silence rule necessary to stop the host handing out
/// its friends' castles five seconds into every match.
///
/// "The picture has not moved" is the thing both readers actually want,
/// and it is not a matter of timing within a frame.
#[derive(Default)]
struct StallWatch {
    /// Seconds the round has been unable to move, and the frame it has
    /// been unable to move off.
    stalled_for: f32,
    stalled_on: u32,
    /// Who the status line is naming, and how long it has left to keep
    /// naming them.
    ///
    /// A latch rather than a reading of `stalled_for`, because that number
    /// snaps back to zero the moment one frame gets through, and a round
    /// that is limping gets a frame through all the time. Reading it
    /// directly put the line on and off once a second: a strobe in the
    /// corner of the eye, which is worse than saying nothing at all.
    waiting_on: Option<u8>,
    waiting_hold: f32,
}

impl StallWatch {
    /// Nothing is stuck and nobody is being waited for, as of `frame`: a
    /// new round, or a round holding still on purpose.
    fn reset(&mut self, frame: u32) {
        let Self {
            stalled_for,
            stalled_on,
            waiting_on,
            waiting_hold,
        } = self;
        *stalled_for = 0.0;
        *stalled_on = frame;
        *waiting_hold = 0.0;
        *waiting_on = None;
    }
}

/// The peers' state hashes against our own.
#[derive(Default)]
struct HashCheck {
    /// First frame where the peers' state hashes disagreed, if ever: a
    /// determinism bug surfaced loudly rather than played through.
    desync_at: Option<u32>,
    own: VecDeque<(u32, u64)>,
    /// Peer hashes for frames we may not have simulated yet; compared as
    /// our own hashes appear, so no exchanged check is skipped.
    peers: VecDeque<(u32, u64)>,
}

impl HashCheck {
    /// Our own hash for `frame`. A short window: the peers are at most a
    /// commit lead behind.
    fn record_own(&mut self, frame: u32, hash: u64) {
        self.own.push_back((frame, hash));
        while self.own.len() > 16 {
            self.own.pop_front();
        }
    }

    /// A peer's hash for `frame`, which we may not have simulated yet.
    fn record_peer(&mut self, frame: u32, hash: u64) {
        self.peers.push_back((frame, hash));
        while self.peers.len() > 64 {
            self.peers.pop_front();
        }
    }

    /// Compare what has been exchanged, and remember the first frame that
    /// disagreed.
    fn compare(&mut self) {
        for &(frame, peer) in &self.peers {
            if let Some(&(_, own)) = self.own.iter().find(|(f, _)| *f == frame)
                && own != peer
            {
                self.desync_at.get_or_insert(frame);
            }
        }
    }

    /// Last round's disagreement is last round's; a new board starts
    /// level, and the hashes that proved it are gone with it.
    fn reset(&mut self) {
        *self = HashCheck::default();
    }
}

/// The goodbye a hosted beach owes the network when its session ends.
///
/// A hosted beach stops being announced when its session goes, whichever
/// way that happens: the match ending, the player quitting to the menu, a
/// desync giving up. Holding the duty here, beside the announcer itself,
/// rather than at each of those exits is what keeps the promise: there is
/// no path that drops an announcing session without either saying goodbye
/// or deliberately taking the announcer back out through
/// [`OnlineSession::back_to_the_lobby`], the one exit where the beach is
/// not going anywhere.
struct Farewell {
    /// `None` on a joiner and on the direct pair, which never announced,
    /// and in the shell a session leaves behind when its beach goes home.
    announcer: Option<Announcer>,
    /// The game port the beacons named, taken when the announcer came
    /// aboard: the port never changes after the bind, and it has to be
    /// readable at drop time.
    port: u16,
}

impl Farewell {
    /// A session with nothing to announce and so nothing to owe.
    fn silent() -> Farewell {
        Farewell {
            announcer: None,
            port: 0,
        }
    }

    /// The announcer, while one is aboard.
    fn aboard(&self) -> Option<&Announcer> {
        self.announcer.as_ref()
    }

    /// Take the announcer back out, disarming the goodbye: the beach is
    /// going back on the air, not away.
    fn disarm(mut self) -> Option<Announcer> {
        self.announcer.take()
    }
}

impl Drop for Farewell {
    fn drop(&mut self) {
        if let Some(announcer) = &self.announcer {
            announcer.closing(self.port);
        }
    }
}

/// What survives of a session when its table walks back to the lobby
/// together at the end of a match: the sockets, still connected, and the
/// bookkeeping the lobby needs to stand the beach back up.
pub struct LobbyReturn {
    /// Host side: the beacon, going back on the air as open. `None` for a
    /// joiner.
    pub announcer: Option<Announcer>,
    pub transport: UdpTransport,
    pub game_name: String,
    /// Host side: every registered peer by its socket index, with the
    /// name its greeting carried and whether it watches, by the same rule
    /// the next round would have seated it. Nobody holds a chair any
    /// more: the lobby deals them again at the launch.
    pub peers: PeerBook,
    /// Joiner side: this peer was at the rail, and greets with `Watch`.
    pub watching: bool,
    pub host: bool,
    /// The seed of the round just played. A host still on its results
    /// card re-answers every greeting with that round's `Start`, and a
    /// lobby that has just walked out of it must read the repeat as
    /// stale rather than as an invitation straight back in.
    pub played_seed: u64,
}

#[derive(Resource, Default)]
pub struct Online(pub Option<OnlineSession>);

/// Pass `msg` from peer `from` on to every other peer. Star topology: the
/// spokes cannot hear each other, so whatever one says reaches the rest
/// only by the hub repeating it, and inputs, pauses, resumes and chat
/// are all repeated this same way.
pub(crate) fn relay(transport: &UdpTransport, from: usize, msg: NetMsg) {
    for other in 0..transport.peer_count() {
        if other != from {
            transport.send_to(other, msg.clone());
        }
    }
}

/// Drain the session while the results card is up. See
/// [`OnlineSession::poll_between_rounds`] for why this cannot wait for the
/// sim: the sim is exactly what has stopped.
pub fn poll_between_rounds(time: Res<Time>, mut online: ResMut<Online>) {
    if let Some(session) = &mut online.0 {
        session.poll_between_rounds(time.delta_secs());
    }
}

mod peers;
mod presence;
mod rounds;
pub(crate) use presence::{abandon_the_departed, leave_a_hostless_round};
pub use rounds::Invitation;
pub(crate) use rounds::lockstep_for;

impl OnlineSession {
    pub fn new(
        transport: UdpTransport,
        session: Lockstep,
        seats: u8,
        terms: MatchTerms,
    ) -> OnlineSession {
        OnlineSession {
            transport,
            session,
            seats,
            terms,
            beach: Vec::new(),
            peers: PeerBook::default(),
            names: Default::default(),
            hashes: HashCheck::default(),
            stall: StallWatch::default(),
            abandoned: Vec::new(),
            home: Home::nowhere(),
            next_round: false,
            series_standing: None,
            resume_echo: 0,
        }
    }

    /// Keep the beach on the air while the round runs, so a lobby can list
    /// it as in progress rather than not at all. Hosts only, and the beacon
    /// says running, so nobody is offered a join that lockstep could not
    /// honour; what it offers is a place in the queue.
    ///
    /// `taken` counts the humans, not the table: an AI seat gives way to a
    /// player who wants it, the same way the lobby fills bots in behind
    /// whoever turned up. The humans in line for the next round count
    /// too, since the chair each will take is spoken for: the lobby
    /// refuses a beach with no chair "this round or the next", and a
    /// beacon that left the queue out let a sixth player in to be dealt
    /// the rail with no notice.
    pub fn keep_announcing(&mut self, delta: f32) {
        if self.home.announcer.aboard().is_none()
            || !crate::app::lobby::once_a_second(&mut self.home.announce_in, delta)
        {
            return;
        }
        let taken = self.players_spoken_for();
        let (Some(announcer), Ok(addr)) =
            (self.home.announcer.aboard(), self.transport.local_addr())
        else {
            return;
        };
        announcer.running(
            addr.port(),
            crate::transport::OnAir {
                name: &self.home.game_name,
                // Seat zero is the host's, and the host is the only one that
                // announces.
                host: &self.names[0],
                taken,
                seats: MAX_PLAYERS as u8,
            },
        );
    }

    /// Host: carry the beacon out of the lobby, and with it the goodbye
    /// owed to the network from now on.
    pub fn stay_on_air(&mut self, announcer: Announcer) {
        // Port 0 if the socket will not say: the goodbye is matched by
        // the announcer's id first, so it still clears the right row.
        let port = self
            .transport
            .local_addr()
            .map(|addr| addr.port())
            .unwrap_or(0);
        self.home.announcer = Farewell {
            announcer: Some(announcer),
            port,
        };
    }

    /// Dismantle a finished match into what the lobby needs to stand the
    /// beach back up, goodbye disarmed: this table is going back to the
    /// lobby together, not away.
    pub fn back_to_the_lobby(mut self) -> LobbyReturn {
        let host = self.is_host();
        let watching = self.watching();
        // Each peer under the name its chair carried, which is the one
        // every screen showed, and the plan's chairs given up: back in the
        // lobby the plan is dealt again, and only who watches survives it.
        for peer in 0..self.peers.len() {
            if let Some(name) = self.peer_name(peer) {
                let name = name.to_string();
                self.peers.row(peer).name = name;
            }
        }
        self.peers.unseat();
        let Self {
            transport,
            session: _,
            seats: _,
            terms,
            beach: _,
            peers,
            names: _,
            hashes: _,
            stall: _,
            abandoned: _,
            home,
            next_round: _,
            series_standing: _,
            resume_echo: _,
        } = self;
        let Home {
            announcer,
            game_name,
            announce_in: _,
            from_lobby: _,
            greet_in: _,
        } = home;
        LobbyReturn {
            announcer: announcer.disarm(),
            transport,
            game_name,
            peers,
            watching,
            host,
            played_seed: terms.seed,
        }
    }

    /// The first frame the peers' state hashes disagreed on, if ever.
    pub fn desync_at(&self) -> Option<u32> {
        self.hashes.desync_at
    }

    /// Call a pause, and tell the peers which frame it lands on. A
    /// spectator's Escape opens their own menu; the match plays on behind
    /// it, and they are still in step when they close it, so the lockstep
    /// names no frame and nothing is sent.
    pub fn request_pause(&mut self) {
        if let Some(frame) = self.session.request_pause() {
            self.transport.send(NetMsg::Pause { frame });
        }
    }

    /// Play on, and keep saying so for a moment (see `RESUME_ECHOES`).
    pub fn request_resume(&mut self) {
        if self.watching() {
            return;
        }
        let frame = self.session.resume();
        self.resume_echo = RESUME_ECHOES;
        self.transport.send(NetMsg::Resume { frame });
    }

    /// The `Resume` to repeat: the pause most recently lifted here.
    fn resume_msg(&self) -> NetMsg {
        NetMsg::Resume {
            frame: self.session.lifted_pause().unwrap_or(0),
        }
    }

    /// Is this session the star's hub? Seat 0 hosts and relays.
    ///
    /// Public because the shell has to know who calls the next round, which
    /// it did through a second method of the same one line, under a second
    /// copy of this sentence.
    pub fn is_host(&self) -> bool {
        self.session.seat() == Some(0)
    }

    /// Whether this peer watches rather than plays.
    pub fn watching(&self) -> bool {
        self.session.watching()
    }

    /// The humans at the table and in line for it: the players in the
    /// lockstep plus every queued peer that did not ask to watch. What the
    /// running beacon reports as taken.
    pub(crate) fn players_spoken_for(&self) -> u8 {
        let queued = self
            .peers
            .iter()
            .filter(|peer| peer.place == peers::Place::Queued && !peer.watch)
            .count();
        (self.session.player_count() + queued).min(MAX_PLAYERS) as u8
    }

    /// One fixed-tick's worth of network pumping: commit and (re)send local
    /// input, drain the socket (the host relays joiner inputs to the other
    /// joiners and re-answers stray hellos with their seat), then simulate
    /// every frame that has complete inputs via `tick`. Records hashes on
    /// the [`HASH_INTERVAL`] cadence. Returns whether the local action was
    /// committed (false = at the commit lead; retry it next tick).
    pub fn pump(&mut self, local_action: PlayerAction, mut tick: impl FnMut(&mut Self)) -> bool {
        let committed = self.session.commit_local(local_action).is_some();
        for &msg in self.session.recent_commits() {
            self.transport.send(NetMsg::Input(msg));
        }
        // Pause state is repeated every tick rather than sent once: UDP
        // drops, and the peer that misses a Pause would otherwise sit
        // watching a frozen beach with no card, while a missed Resume would
        // leave the session stopped for good. Both self-heal here: the
        // resume repeats until frames actually start moving again.
        match self.session.pause_frame() {
            Some(frame) => self.transport.send(NetMsg::Pause { frame }),
            None if self.resume_echo > 0 => {
                self.resume_echo -= 1;
                self.transport.send(self.resume_msg());
            }
            None => {}
        }
        let host = self.is_host();
        // Set when the host's call for the next round arrives in this very
        // pump, which is possible for a joiner or spectator still on the
        // last frames of the round the host has finished. The lockstep is
        // fresh from frame zero then, and the board is still the old one:
        // ticking would play the new round's first frames out on it, and
        // the joiner would enter the arena two frames ahead of every other
        // table, a desync at the first hash.
        let mut rearmed = false;
        // Likewise every seat given up on, for the rest of the round: a
        // joiner that misses the one notice waits on that seat forever,
        // and the host never notices, because a joiner that is waiting is
        // still talking. Five bytes a seat a tick.
        if host {
            for &(seat, frame) in &self.abandoned {
                self.transport.send(NetMsg::Abandoned { seat, frame });
            }
        }
        for (msg, from) in self.transport.recv_all() {
            // Whatever it was, that peer is still there, which is all
            // `abandon_stalled` and `host_gone` go on.
            self.mark_heard(from);
            match msg {
                // A peer that missed the Start datagram keeps greeting us;
                // repeat what it is until it stops. A watcher is told it is
                // watching, so the seat it gets is not one.
                NetMsg::Hello { name } => {
                    if host {
                        let answer = self.answer_greeting(from, &name_from_wire(&name), false);
                        self.transport.send_to(from, answer);
                    }
                }
                NetMsg::Watch => {
                    if host {
                        // Watching is no more possible than playing once the
                        // round is under way: a spectator simulates the same
                        // frames as everyone else, from the same frame zero.
                        // The wish is remembered, so a peer that armed W and
                        // dialled in mid-round is seated as an onlooker next
                        // round rather than dealt a chair it never asked for.
                        let answer = self.answer_greeting(from, "", true);
                        self.transport.send_to(from, answer);
                    }
                }
                NetMsg::Input(input) => {
                    // The host knows which seat each peer was given, and
                    // takes inputs for that seat alone: a peer speaking for
                    // another's seat (a bug, or a spectator with ideas)
                    // would otherwise be believed by whoever heard it
                    // first, and the seats would diverge. A joiner hears
                    // only the host, which has already done this. The
                    // direct `PINCH_HOST` pair keeps no plan, having never
                    // been through a lobby; with nothing to check against
                    // it takes any seat but its own, as it always did.
                    if host {
                        let allowed = if self.peers.planned() == 0 {
                            Some(input.player) != self.session.seat()
                        } else {
                            self.seat_of(from) == Some(input.player)
                        };
                        if !allowed {
                            continue;
                        }
                    }
                    self.session.receive(input);
                    if host {
                        relay(&self.transport, from, NetMsg::Input(input));
                    }
                }
                NetMsg::Hash { frame, hash } => {
                    // The peer may be ahead of us; buffer until we simulate
                    // that frame ourselves.
                    self.hashes.record_peer(frame, hash);
                }
                NetMsg::Pause { frame } => {
                    self.session.receive_pause(frame);
                    if host {
                        relay(&self.transport, from, NetMsg::Pause { frame });
                    }
                }
                NetMsg::Resume { frame } => {
                    let frame = self.session.receive_resume(frame);
                    self.resume_echo = RESUME_ECHOES;
                    if host {
                        relay(&self.transport, from, NetMsg::Resume { frame });
                    }
                }
                NetMsg::Start {
                    seats,
                    seat,
                    terms,
                    names,
                    standing,
                    beach,
                } if !host && self.is_next_round(&terms) => {
                    // The host has called the next round. A fresh seed is
                    // what says so; everything else about the round may
                    // well be the same. Taking it up here only rearms the
                    // session; the board is rebuilt on the way back into
                    // the arena, by the same path a first round uses.
                    self.take_up(Invitation {
                        seats,
                        seat,
                        terms,
                        names,
                        standing,
                        beach,
                    });
                    self.next_round = true;
                    rearmed = true;
                }
                NetMsg::Start { names, .. } => {
                    // Stale as a launch signal, but a joiner in the direct
                    // PINCH_HOST/PINCH_JOIN pair, which never waits in a
                    // lobby, learns the table's names from it.
                    if !host {
                        for (slot, name) in self.names.iter_mut().zip(table_from_wire(&names)) {
                            if !name.is_empty() {
                                *slot = name;
                            }
                        }
                    }
                }
                // Someone on another build is talking to this port. Every
                // peer in a running match paired before it started, so this
                // is a stranger, not a member: the match plays on without it.
                NetMsg::Incompatible { .. } => {}
                // Only a host hands these out, and a host never receives
                // one: a joiner in the queue is still in the lobby.
                NetMsg::Queued { .. } => {}
                // Chat is a lobby thing. A line that arrives mid-round is
                // from a peer still sitting on its lobby screen, and there
                // is nowhere here to show it.
                NetMsg::Chat { .. } => {}
                NetMsg::Abandoned { seat, frame } => {
                    // The host's word, not our own patience. Idempotent,
                    // because it is repeated against packet loss.
                    if !host && !self.abandoned.iter().any(|(gone, _)| *gone == seat) {
                        self.session.abandon(seat, frame);
                        self.abandoned.push((seat, frame));
                    }
                }
                // The lobby's business, and the lobby is behind us.
                NetMsg::Roster { .. } => {}
            }
        }
        self.hashes.compare();
        if rearmed {
            return committed;
        }
        // At most a couple of frames per fixed tick: a lagging peer catches
        // up gradually instead of spiralling.
        for _ in 0..2 {
            tick(self);
        }
        self.hashes.compare();
        committed
    }

    /// Called by the tick closure after simulating a frame, with the fresh
    /// state hash.
    pub fn after_frame(&mut self, hash: u64) {
        let frame = self.session.frame();
        if frame.is_multiple_of(HASH_INTERVAL) {
            self.hashes.record_own(frame, hash);
            self.transport.send(NetMsg::Hash { frame, hash });
        }
    }
}

/// Parse the dev env vars into a ready session, if requested.
/// `PINCH_HOST=47777` or `PINCH_JOIN=192.168.1.10:47777`.
pub fn session_from_env() -> Option<OnlineSession> {
    // The direct hooks skip the lobby, so there is no Start to agree on:
    // both processes must be told the same terms. `PINCH_BOTS=n` seats n AI
    // players behind the two humans and has to be set on both sides.
    let bots: u8 = crate::app::dev::bots().unwrap_or(0);
    // Two humans and the AI behind them; MAX_PLAYERS is the ceiling, and
    // the clamp is on the bots so a wild PINCH_BOTS cannot wrap the sum
    // back under the two humans.
    let humans = 2u8;
    let bots = bots.min(crate::sim::MAX_PLAYERS as u8 - humans);
    let seats = humans + bots;
    let terms = MatchTerms {
        bots,
        ..crate::app::match_setup::terms(
            &crate::app::match_setup::MatchConfig::default(),
            crate::app::teams::TeamMode::Solo,
            0,
        )
    };
    // The name this player goes by online: their P1 name from settings,
    // read straight off disk because the dev hooks run before the app's
    // resources are wired up.
    let own = crate::app::settings::GameSettings::load().0.names[0].clone();
    if let Some(port) = crate::app::dev::direct_host() {
        let port: u16 = port.parse().ok()?;
        let transport = UdpTransport::host(port).ok()?;
        info!("hosting on UDP port {port}, waiting for a peer ({seats} seats)");
        let mut session = OnlineSession::new(
            transport,
            Lockstep::new(0, vec![0, 1], crate::sim::DEFAULT_DELAY),
            seats,
            terms,
        );
        session.names[0] = own;
        return Some(session);
    }
    if let Some(addr) = crate::app::dev::direct_join() {
        let transport = UdpTransport::join(addr.as_str()).ok()?;
        transport.send(NetMsg::hello(&own));
        info!("joining {addr} ({seats} seats)");
        let mut session = OnlineSession::new(
            transport,
            Lockstep::new(1, vec![0, 1], crate::sim::DEFAULT_DELAY),
            seats,
            terms,
        );
        session.names[1] = own;
        return Some(session);
    }
    None
}

#[cfg(test)]
mod homecoming_tests {
    use super::*;
    use crate::sim::DEFAULT_DELAY;
    use crate::transport::{Beacon, Discovery};

    fn hosting_session(seed: u64) -> OnlineSession {
        OnlineSession::new(
            UdpTransport::host(0).expect("game socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            MatchTerms {
                seed,
                ..MatchTerms::default()
            },
        )
    }

    /// Beacons from other tests share the machine, so every packet is
    /// judged by whether it names our game port.
    fn heard_for(discovery: &mut Discovery, port: u16) -> Vec<Beacon> {
        let mut heard = Vec::new();
        for _ in 0..20 {
            std::thread::sleep(std::time::Duration::from_millis(10));
            heard.extend(
                discovery
                    .poll()
                    .into_iter()
                    .filter(|(addr, _)| addr.port() == port)
                    .map(|(_, beacon)| beacon),
            );
            if !heard.is_empty() {
                break;
            }
        }
        heard
    }

    /// The goodbye still fires on every ordinary way out - quitting to the
    /// menu, a desync giving up, the process letting go - which is the
    /// promise the [`Farewell`] guard carries for the whole session.
    #[test]
    fn a_dropped_session_still_says_goodbye() {
        let mut discovery = Discovery::bind().expect("lobby port");
        let mut session = hosting_session(0);
        let port = session.transport.local_addr().expect("addr").port();
        session.stay_on_air(Announcer::new(0xD00D).expect("announcer"));
        drop(session);
        let heard = heard_for(&mut discovery, port);
        let beacon = heard.first().expect("the goodbye went out");
        assert!(matches!(beacon, Beacon::Closing { .. }), "{beacon:?}");
    }

    /// The one exit where the beach is not going anywhere: walking the
    /// table back to the lobby keeps the socket, keeps the beacon, and
    /// says no goodbye, because the same beach is about to announce
    /// itself open again.
    #[test]
    fn the_way_back_to_the_lobby_says_no_goodbye() {
        let mut discovery = Discovery::bind().expect("lobby port");
        let mut session = hosting_session(42);
        let port = session.transport.local_addr().expect("addr").port();
        session.stay_on_air(Announcer::new(0xB0A7).expect("announcer"));
        session.home.game_name = "Room 3".into();
        session.home.from_lobby = true;
        let returned = session.back_to_the_lobby();
        assert!(returned.host);
        assert!(!returned.watching);
        assert_eq!(returned.played_seed, 42, "the round it walked out of");
        assert_eq!(returned.game_name, "Room 3");
        assert_eq!(
            returned.transport.local_addr().expect("addr").port(),
            port,
            "the same socket every peer is still talking to"
        );
        assert!(
            returned.announcer.is_some(),
            "the beacon rides home to announce again"
        );
        std::thread::sleep(std::time::Duration::from_millis(40));
        assert!(
            discovery
                .poll()
                .into_iter()
                .all(|(addr, _)| addr.port() != port),
            "no goodbye for a beach that is coming home"
        );
    }

    /// What the lobby gets back to seat people with: each peer's name as
    /// its greeting carried it, and the watchers by the same rule the next
    /// round would have used - a spectator's `None` in the launch plan, or
    /// a watch wish greeted mid-round.
    #[test]
    fn the_table_walks_back_with_its_names_and_watchers() {
        let mut session = hosting_session(0);
        let port = session.transport.local_addr().expect("addr").port();
        // Two peers, registered one at a time so their indices are known:
        // Bo plays, and somebody nameless watches from the rail.
        let bo = UdpTransport::join(("127.0.0.1", port)).expect("join");
        bo.send(NetMsg::hello("Bo"));
        let rail = UdpTransport::join(("127.0.0.1", port)).expect("join");
        for want in [1, 2] {
            if want == 2 {
                rail.send(NetMsg::Watch);
            }
            for _ in 0..40 {
                std::thread::sleep(std::time::Duration::from_millis(5));
                for (msg, from) in session.transport.recv_all() {
                    match msg {
                        NetMsg::Hello { name } => {
                            let told = name_from_wire(&name);
                            session.remember_peer_name(from, &told);
                        }
                        NetMsg::Watch => session.note_watch_wish(from),
                        NetMsg::Input(_)
                        | NetMsg::Hash { .. }
                        | NetMsg::Start { .. }
                        | NetMsg::Pause { .. }
                        | NetMsg::Resume { .. }
                        | NetMsg::Queued { .. }
                        | NetMsg::Chat { .. }
                        | NetMsg::Roster { .. }
                        | NetMsg::Abandoned { .. }
                        | NetMsg::Incompatible { .. } => {}
                    }
                }
                if session.transport.peer_count() >= want {
                    break;
                }
            }
            assert_eq!(session.transport.peer_count(), want, "peer registered");
        }
        session.peers.deal(&[Some(1), None]);
        let returned = session.back_to_the_lobby();
        let names: Vec<&str> = returned.peers.iter().map(|p| p.name.as_str()).collect();
        assert_eq!(
            names,
            ["Bo", ""],
            "each chair keeps the name its greeting carried"
        );
        let watching: Vec<bool> = returned.peers.iter().map(Peer::watches).collect();
        assert_eq!(watching, [false, true], "and the rail stays the rail");
    }

    /// A joiner's way back: no beacon to carry, but the socket, the seed
    /// it played on, and whether it was watching all come home with it.
    #[test]
    fn a_watching_joiner_comes_home_to_the_rail() {
        let session = OnlineSession::new(
            UdpTransport::join(("127.0.0.1", 47999)).expect("join"),
            Lockstep::observer(vec![0, 1], DEFAULT_DELAY),
            2,
            MatchTerms {
                seed: 7,
                ..MatchTerms::default()
            },
        );
        let returned = session.back_to_the_lobby();
        assert!(!returned.host);
        assert!(returned.watching, "the rail is remembered");
        assert!(returned.announcer.is_none(), "a joiner has no beacon");
        assert_eq!(returned.played_seed, 7);
    }
}

#[cfg(test)]
mod table_tests {
    use super::*;
    use crate::sim::DEFAULT_DELAY;

    fn session(local: Option<u8>, port: u16) -> OnlineSession {
        let transport = match local {
            Some(0) => UdpTransport::host(0).expect("host socket"),
            _ => UdpTransport::join(("127.0.0.1", port)).expect("join"),
        };
        OnlineSession::new(
            transport,
            match local {
                Some(seat) => Lockstep::new(seat, vec![0, 1], DEFAULT_DELAY),
                None => Lockstep::observer(vec![0, 1], DEFAULT_DELAY),
            },
            2,
            MatchTerms::default(),
        )
    }

    /// The running beacon says how many chairs are spoken for: the humans
    /// in the lockstep and everyone in line who did not ask to watch. It
    /// used to count the lockstep alone, so with two playing and four
    /// queued it still said two of six.
    #[test]
    fn the_beacon_counts_the_queue_as_taken() {
        let mut host = session(Some(0), 0);
        assert_eq!(host.players_spoken_for(), 2, "the two at the table");
        host.peers.row(0).name = "Bo".into();
        assert_eq!(host.players_spoken_for(), 3, "one in line for a chair");
        host.peers.row(1).watch = true;
        assert_eq!(host.players_spoken_for(), 3, "a watcher wants no chair");
        for peer in 2..8 {
            host.peers.row(peer).name = "Cy".into();
        }
        assert_eq!(
            host.players_spoken_for(),
            MAX_PLAYERS as u8,
            "and never more chairs than the table has"
        );
    }

    /// A joiner reads its own seat in the host's abandonment notice: that
    /// is how it learns it was dropped, rather than waiting out the host
    /// and being told the host had left. Somebody else's seat, or the
    /// host's own view, says nothing of the kind.
    #[test]
    fn a_seat_told_of_its_own_abandonment_knows_it_was_dropped() {
        let host = session(Some(0), 0);
        let port = host.transport.local_addr().expect("addr").port();
        let mut joiner = session(Some(1), port);
        assert!(!joiner.dropped(), "nothing said yet");
        joiner.abandoned.push((0, 30));
        assert!(!joiner.dropped(), "another seat's abandonment is theirs");
        joiner.abandoned.push((1, 30));
        assert!(joiner.dropped(), "its own is the word it was waiting for");
        let mut host = host;
        host.abandoned.push((0, 30));
        assert!(!host.dropped(), "the host drops nobody it is");
        let mut watcher = session(None, port);
        watcher.abandoned.push((1, 30));
        assert!(
            !watcher.dropped(),
            "a spectator holds no seat to be dropped from"
        );
    }
}

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

    /// The desync check has to work across the gap between two peers, and
    /// the gap runs both ways: a peer a commit lead ahead of us sends its
    /// hash for a frame we have not simulated yet.
    ///
    /// So a peer's hash is kept rather than compared on arrival and
    /// weighed once our own for that frame appears. Comparing only what
    /// was already in hand would have skipped exactly the checks a lagging
    /// peer sends, which is the peer a desync is most likely to be about.
    #[test]
    fn a_peer_hash_that_arrives_early_is_still_weighed() {
        let mut hashes = HashCheck::default();
        // It arrives first, and there is nothing yet to hold it against.
        hashes.record_peer(60, 0xBAD_B0A7);
        hashes.compare();
        assert_eq!(hashes.desync_at, None, "nothing to compare it to yet");
        // A frame in between, which both sides agree on, is not a desync.
        hashes.record_own(30, 0xC0FFEE);
        hashes.record_peer(30, 0xC0FFEE);
        hashes.compare();
        assert_eq!(hashes.desync_at, None, "and they agree about frame 30");
        // We reach 60 at last, and the hash that has been waiting is what
        // catches it.
        hashes.record_own(60, 0x600D_B0A7);
        hashes.compare();
        assert_eq!(hashes.desync_at, Some(60), "the waiting hash caught it");
    }

    /// A round that agreed all the way through never raises one, however
    /// many hashes crossed: the check is loud, and a loud check that cries
    /// on a healthy round is one nobody believes on a sick one.
    #[test]
    fn a_round_that_agrees_is_never_flagged_and_a_new_one_starts_level() {
        let mut hashes = HashCheck::default();
        for frame in (0..300).step_by(HASH_INTERVAL as usize) {
            let hash = u64::from(frame) * 0x9E37_79B9;
            hashes.record_own(frame, hash);
            hashes.record_peer(frame, hash);
            hashes.compare();
        }
        assert_eq!(hashes.desync_at, None, "every frame agreed");

        // Last round's disagreement is last round's. A new board starts
        // level, or one bad round would paint every round after it.
        hashes.record_own(30, 1);
        hashes.record_peer(30, 2);
        hashes.compare();
        assert_eq!(hashes.desync_at, Some(30));
        hashes.reset();
        hashes.compare();
        assert_eq!(hashes.desync_at, None, "and the new board is level");
    }
}