pinch-points 0.1.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, UdpTransport, name_from_wire, wire_name};
use bevy::prelude::*;
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>,
    /// Host side: the seat each peer index was given, `None` for the
    /// watchers among them. Empty on a joiner, which answers nobody.
    pub peer_seats: Vec<Option<u8>>,
    /// Host side: what each peer index calls itself, from its greeting,
    /// and whether it asked to watch rather than play. Kept for the peers
    /// the launch plan does not cover: someone who queued mid-round holds
    /// no seat yet, so its name and its wish to watch have nowhere else to
    /// live until [`Self::call_next_round`] seats it. Grown as peers appear
    /// and shifted with the socket's own list by [`Self::forget_peer_row`].
    peer_names: Vec<String>,
    peer_watch: Vec<bool>,
    /// 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],
    /// First frame where the peers' state hashes disagreed, if ever: a
    /// determinism bug surfaced loudly rather than played through.
    pub desync_at: Option<u32>,
    /// Seconds the round has been unable to move, and the frame it has been
    /// unable to move off.
    ///
    /// 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.
    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,
    /// Seconds since each peer index was last heard from: anything at all,
    /// an input, a hash, a stray greeting.
    ///
    /// Kept beside the transport's own peer list and shifted with it. This
    /// 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 entry it has,
    /// the host, and calls the round off when that goes quiet.
    peer_silence: Vec<f32>,
    /// 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)>,
    /// 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. `None` on a joiner,
    /// and on a direct `PINCH_HOST` pair, which never announced at all.
    pub announcer: Option<Announcer>,
    /// 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,
    /// 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,
    /// 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<(u8, [u8; MAX_PLAYERS])>,
    /// Ticks of `Resume` still to repeat (see [`RESUME_ECHOES`]).
    resume_echo: u8,
    own_hashes: VecDeque<(u32, u64)>,
    /// Peer hashes for frames we may not have simulated yet; compared (and
    /// drained) as our own hashes appear, so no exchanged check is skipped.
    peer_hashes: VecDeque<(u32, u64)>,
}

/// 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. Putting it here rather than at each of those exits is
/// what keeps the promise: there is no path that drops a session and
/// forgets to take the beach off the network with it.
impl Drop for OnlineSession {
    fn drop(&mut self) {
        if let Some(announcer) = &self.announcer
            && let Ok(addr) = self.transport.local_addr()
        {
            announcer.closing(addr.port());
        }
    }
}

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

/// 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 presence;
mod rounds;
pub(crate) use presence::{abandon_the_departed, leave_a_hostless_round};

impl OnlineSession {
    pub fn new(
        transport: UdpTransport,
        session: Lockstep,
        seats: u8,
        terms: MatchTerms,
    ) -> OnlineSession {
        OnlineSession {
            transport,
            session,
            seats,
            terms,
            beach: Vec::new(),
            peer_seats: Vec::new(),
            peer_names: Vec::new(),
            peer_watch: Vec::new(),
            names: Default::default(),
            stalled_for: 0.0,
            stalled_on: 0,
            waiting_on: None,
            waiting_hold: 0.0,
            peer_silence: Vec::new(),
            abandoned: Vec::new(),
            announcer: None,
            game_name: String::new(),
            announce_in: 0.0,
            greet_in: 0.0,
            next_round: false,
            series_standing: None,
            desync_at: None,
            resume_echo: 0,
            own_hashes: VecDeque::new(),
            peer_hashes: VecDeque::new(),
        }
    }

    /// 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.
    pub fn keep_announcing(&mut self, delta: f32) {
        if self.announcer.is_none() {
            return;
        }
        self.announce_in -= delta;
        if self.announce_in > 0.0 {
            return;
        }
        self.announce_in = crate::app::lobby::ANNOUNCE_EVERY;
        let taken = self.session.player_count() as u8;
        let (Some(announcer), Ok(addr)) = (&self.announcer, self.transport.local_addr()) else {
            return;
        };
        announcer.running(
            addr.port(),
            crate::transport::OnAir {
                name: &self.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,
            },
        );
    }

    /// Call a pause, and tell the peers which frame it lands on.
    pub fn request_pause(&mut self) {
        if self.watching() {
            // A spectator's Escape opens their own menu; the match plays on
            // behind it, and they are still in step when they close it.
            return;
        }
        let 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()
    }

    /// 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();
        // 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 {
                        // Its name is written down whether it is at the
                        // table or waiting in line: a peer that queues
                        // mid-round holds no seat to keep it in yet, and
                        // without this it was seated nameless next round.
                        let told = name_from_wire(&name);
                        if !told.is_empty() {
                            self.remember_peer_name(from, &told);
                        }
                        if let Some(queued) = self.queue_place(from) {
                            self.transport.send_to(from, queued);
                            continue;
                        }
                        let seat = self.seat_of(from);
                        if let Some(slot) =
                            seat.and_then(|seat| self.names.get_mut(usize::from(seat)))
                            && !told.is_empty()
                        {
                            *slot = told;
                        }
                        let start = self.start_msg(seat);
                        self.transport.send_to(from, start);
                    }
                }
                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.
                        self.note_watch_wish(from);
                        let answer = self
                            .queue_place(from)
                            .unwrap_or_else(|| self.start_msg(None));
                        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.peer_seats.is_empty() {
                            Some(input.player) != self.session.seat()
                        } else {
                            self.seat_of(from) == Some(input.player)
                        };
                        if !allowed {
                            continue;
                        }
                    }
                    self.session.receive(input);
                    if host {
                        // Star topology: joiners only see the host, so the
                        // host forwards every input to the other joiners.
                        for other in 0..self.transport.peer_count() {
                            if other != from {
                                self.transport.send_to(other, NetMsg::Input(input));
                            }
                        }
                    }
                }
                NetMsg::Hash { frame, hash } => {
                    // The peer may be ahead of us; buffer until we simulate
                    // that frame ourselves.
                    self.peer_hashes.push_back((frame, hash));
                    while self.peer_hashes.len() > 64 {
                        self.peer_hashes.pop_front();
                    }
                }
                NetMsg::Pause { frame } => {
                    self.session.receive_pause(frame);
                    if host {
                        // Star topology again: only the host hears everyone,
                        // so it passes the pause on to the other joiners.
                        for other in 0..self.transport.peer_count() {
                            if other != from {
                                self.transport.send_to(other, NetMsg::Pause { frame });
                            }
                        }
                    }
                }
                NetMsg::Resume { frame } => {
                    let frame = self.session.receive_resume(frame);
                    self.resume_echo = RESUME_ECHOES;
                    if host {
                        for other in 0..self.transport.peer_count() {
                            if other != from {
                                self.transport.send_to(other, NetMsg::Resume { frame });
                            }
                        }
                    }
                }
                NetMsg::Start {
                    seats,
                    seat,
                    terms,
                    names,
                    round,
                    wins,
                    beach,
                } if !host && self.is_next_round(&terms) => {
                    self.beach = beach;
                    // 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.
                    let table = std::array::from_fn(|i| name_from_wire(&names[i]));
                    self.begin_round(seats, seat, terms, table);
                    self.series_standing = (terms.series == 1).then_some((round, wins));
                    self.next_round = 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, wire) in self.names.iter_mut().zip(&names) {
                            let name = name_from_wire(wire);
                            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.compare_hashes();
        // 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.compare_hashes();
        committed
    }

    fn compare_hashes(&mut self) {
        for &(frame, peer) in &self.peer_hashes {
            if let Some(&(_, own)) = self.own_hashes.iter().find(|(f, _)| *f == frame)
                && own != peer
            {
                self.desync_at.get_or_insert(frame);
            }
        }
    }

    /// 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.own_hashes.push_back((frame, hash));
            while self.own_hashes.len() > 16 {
                self.own_hashes.pop_front();
            }
            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().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
}