Grovewright/core/src/host.rs

//! The host's side of a shared world: the world, its recording and the
//! guests connected to it.
//!
//! The host runs the ticks. Its own player's actions and the actions its
//! guests send become commands of the next tick, and every tick goes out to
//! every accepted guest. A solo game is a host without guests.

use std::collections::BTreeMap;

use crate::command::Command;
use crate::player::{clamp_name, PlayerId, PlayerKey, MAX_KNOWN_PLAYERS, MAX_ONLINE_PLAYERS};
use crate::protocol::{Action, Refusal, ToGuest, ToHost, CHECK_INTERVAL_TICKS};
use crate::replay::Recording;
use crate::rules::rules_id;
use crate::world::{Event, World};

/// Identifies a connection. The shell chooses the numbers.
pub type Peer = u32;

/// Most actions one player may take per tick. Further actions in the same
/// tick are dropped, so that no guest can flood the world with commands.
pub const MAX_ACTIONS_PER_TICK: u32 = 4;

/// A guest reported that its state differs from the host's checksum.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Divergence {
    pub peer: Peer,
    /// The guest's tick when it noticed.
    pub tick: u64,
    /// The hash of the guest's state at that tick.
    pub hash: u64,
}

/// Where the connection of a guest stands.
#[derive(Clone, Debug, PartialEq, Eq)]
enum Link {
    /// Connected, but the guest has not greeted yet.
    Greeting,
    /// Accepted. The guest has the world, and its join runs in the next tick.
    Joining { key: PlayerKey },
    /// In the world as `player`. `actions` counts its actions in this tick.
    /// `held` is the newest action that arrived while the player's cooldown
    /// still ran; it goes into the first tick after the cooldown.
    Playing {
        player: PlayerId,
        actions: u32,
        held: Option<Action>,
    },
}

/// The host of a shared world.
pub struct Host {
    world: World,
    recording: Recording,
    rules: u64,
    local_key: Option<PlayerKey>,
    local_actions: u32,
    /// Commands for the next tick, in the order they arrived.
    inbox: Vec<Command>,
    links: BTreeMap<Peer, Link>,
    outbox: Vec<(Peer, Vec<u8>)>,
    closing: Vec<Peer>,
    divergences: Vec<Divergence>,
}

impl Host {
    /// Hosts `world`, new or loaded. Nobody is in it yet.
    pub fn new(world: World) -> Self {
        Self {
            recording: Recording::start(&world),
            world,
            rules: rules_id(),
            local_key: None,
            local_actions: 0,
            inbox: Vec::new(),
            links: BTreeMap::new(),
            outbox: Vec::new(),
            closing: Vec::new(),
            divergences: Vec::new(),
        }
    }

    pub fn world(&self) -> &World {
        &self.world
    }

    /// The recording of everything that happened since the world was hosted.
    pub fn recording(&self) -> &Recording {
        &self.recording
    }

    /// Queues the entry of the host's own player for the next tick.
    pub fn join_local(&mut self, key: PlayerKey, name: &str) {
        self.local_key = Some(key);
        self.inbox.push(Command::Join {
            key,
            name: clamp_name(name),
        });
    }

    /// The host's own player, once it is in the world.
    pub fn local_player(&self) -> Option<PlayerId> {
        online_player(&self.world, &self.local_key?)
    }

    /// Queues an action of the host's own player for the next tick. Does
    /// nothing before the player is in the world.
    pub fn act_local(&mut self, action: Action) {
        let Some(player) = self.local_player() else {
            return;
        };
        if self.local_actions < MAX_ACTIONS_PER_TICK {
            self.local_actions += 1;
            self.inbox.push(action.command(player));
        }
    }

    /// A guest connected as `peer`. Its first message must be a greeting.
    pub fn connect(&mut self, peer: Peer) {
        self.links.entry(peer).or_insert(Link::Greeting);
    }

    /// The connection `peer` closed. A guest in the world leaves it in the
    /// next tick; an accepted guest whose join has not run yet never enters.
    pub fn disconnect(&mut self, peer: Peer) {
        match self.links.remove(&peer) {
            Some(Link::Joining { key }) => self.inbox.retain(
                |command| !matches!(command, Command::Join { key: joining, .. } if *joining == key),
            ),
            Some(Link::Playing { player, .. }) => self.inbox.push(Command::Leave { player }),
            Some(Link::Greeting) | None => {}
        }
    }

    /// Handles a message that arrived from `peer`. A guest that breaks the
    /// protocol is closed; see [`Host::take_closing`].
    pub fn receive(&mut self, peer: Peer, bytes: &[u8]) {
        let Some(link) = self.links.get(&peer).cloned() else {
            return;
        };
        match (link, ToHost::decode(bytes)) {
            (Link::Greeting, Ok(ToHost::Hello { rules, key, name })) => {
                self.greet(peer, rules, key, &name);
            }
            (Link::Greeting, Ok(ToHost::ForeignHello { .. })) => {
                self.refuse(peer, Refusal::OtherProtocol);
            }
            (
                Link::Playing {
                    player,
                    actions,
                    held,
                },
                Ok(ToHost::Act(action)),
            ) => {
                if actions < MAX_ACTIONS_PER_TICK {
                    // A guest counts the cooldown in its own ticks, which
                    // lag the host's by the round trip. An action that
                    // arrives too early would be lost, so it waits.
                    let wait = held.is_some() || cooling_down(&self.world, player);
                    if !wait {
                        self.inbox.push(action.command(player));
                    }
                    self.links.insert(
                        peer,
                        Link::Playing {
                            player,
                            actions: actions + 1,
                            held: if wait { Some(action) } else { None },
                        },
                    );
                }
            }
            // The guest acted before its join ran. It cannot know its player
            // yet, so the action is dropped.
            (Link::Joining { .. }, Ok(ToHost::Act(_))) => {}
            (Link::Joining { .. } | Link::Playing { .. }, Ok(ToHost::Diverged { tick, hash })) => {
                self.divergences.push(Divergence { peer, tick, hash });
                self.send(
                    peer,
                    &ToGuest::World {
                        snapshot: self.world.to_snapshot(),
                    },
                );
            }
            _ => {
                self.disconnect(peer);
                self.closing.push(peer);
            }
        }
    }

    fn greet(&mut self, peer: Peer, rules: u64, key: PlayerKey, name: &str) {
        if let Err(refusal) = self.admission(rules, &key) {
            self.refuse(peer, refusal);
            return;
        }
        self.send(
            peer,
            &ToGuest::World {
                snapshot: self.world.to_snapshot(),
            },
        );
        self.inbox.push(Command::Join {
            key,
            name: clamp_name(name),
        });
        self.links.insert(peer, Link::Joining { key });
    }

    /// Whether a guest with `rules` and `key` may join now. Joins that are
    /// queued for the next tick count as if they had run.
    fn admission(&self, rules: u64, key: &PlayerKey) -> Result<(), Refusal> {
        if rules != self.rules {
            return Err(Refusal::OtherRules);
        }
        let queued: Vec<&PlayerKey> = self
            .inbox
            .iter()
            .filter_map(|command| match command {
                Command::Join { key, .. } => Some(key),
                _ => None,
            })
            .collect();
        if online_player(&self.world, key).is_some() || queued.contains(&key) {
            return Err(Refusal::AlreadyPlaying);
        }
        let online = self
            .world
            .players()
            .iter()
            .filter(|player| player.online)
            .count();
        let entering = queued
            .iter()
            .filter(|queued| online_player(&self.world, queued).is_none())
            .count();
        if online + entering >= MAX_ONLINE_PLAYERS {
            return Err(Refusal::WorldFull);
        }
        let new = queued
            .iter()
            .filter(|queued| self.world.find_player(queued).is_none())
            .count();
        if self.world.find_player(key).is_none()
            && self.world.players().len() + new >= MAX_KNOWN_PLAYERS
        {
            return Err(Refusal::TooManyPlayers);
        }
        Ok(())
    }

    fn refuse(&mut self, peer: Peer, refusal: Refusal) {
        self.send(peer, &ToGuest::Refused(refusal));
        self.links.remove(&peer);
        self.closing.push(peer);
    }

    fn send(&mut self, peer: Peer, message: &ToGuest) {
        self.outbox.push((peer, message.encode()));
    }

    /// Runs the next tick with every queued command, sends it to every
    /// accepted guest and returns what happened.
    pub fn step(&mut self) -> Vec<Event> {
        self.release_held_actions();
        let commands = std::mem::take(&mut self.inbox);
        let tick = self.world.tick();
        let events = self.recording.step(&mut self.world, &commands);
        let check = self
            .world
            .tick()
            .is_multiple_of(CHECK_INTERVAL_TICKS)
            .then(|| self.world.state_hash());
        let message = ToGuest::Tick {
            tick,
            commands,
            check,
        }
        .encode();
        for (peer, link) in &self.links {
            if *link != Link::Greeting {
                self.outbox.push((*peer, message.clone()));
            }
        }

        self.local_actions = 0;
        let mut refused = Vec::new();
        for (peer, link) in self.links.iter_mut() {
            match link {
                Link::Joining { key } => match online_player(&self.world, key) {
                    Some(player) => {
                        *link = Link::Playing {
                            player,
                            actions: 0,
                            held: None,
                        }
                    }
                    None => refused.push(*peer),
                },
                Link::Playing { actions, .. } => *actions = 0,
                Link::Greeting => {}
            }
        }
        // The admission check keeps the world from refusing an accepted
        // guest. Should it happen anyway, the guest learns why.
        for peer in refused {
            self.refuse(peer, Refusal::WorldFull);
        }
        events
    }

    /// Queues the held actions whose player's cooldown is over.
    fn release_held_actions(&mut self) {
        for link in self.links.values_mut() {
            let Link::Playing { player, held, .. } = link else {
                continue;
            };
            if held.is_some() && !cooling_down(&self.world, *player) {
                if let Some(action) = held.take() {
                    self.inbox.push(action.command(*player));
                }
            }
        }
    }

    /// Number of accepted guests, including those whose join has not run.
    pub fn guest_count(&self) -> usize {
        self.links
            .values()
            .filter(|link| **link != Link::Greeting)
            .count()
    }

    /// The player of the guest connected as `peer`, once it is in the world.
    pub fn player_of(&self, peer: Peer) -> Option<PlayerId> {
        match self.links.get(&peer) {
            Some(Link::Playing { player, .. }) => Some(*player),
            _ => None,
        }
    }

    /// Messages to send, as (peer, bytes), in the order they were made.
    pub fn take_outbox(&mut self) -> Vec<(Peer, Vec<u8>)> {
        std::mem::take(&mut self.outbox)
    }

    /// Connections the host has given up. The shell closes them after it
    /// has sent the messages for them.
    pub fn take_closing(&mut self) -> Vec<Peer> {
        std::mem::take(&mut self.closing)
    }

    /// Divergences the guests reported since the last call.
    pub fn take_divergences(&mut self) -> Vec<Divergence> {
        std::mem::take(&mut self.divergences)
    }
}

/// Whether `player` still waits for its cooldown in the next tick of
/// `world`, so that an action then would be dropped.
fn cooling_down(world: &World, player: PlayerId) -> bool {
    world
        .player(player)
        .is_some_and(|state| world.tick() < state.next_action_tick)
}

/// The player with `key`, if it is online in `world`.
fn online_player(world: &World, key: &PlayerKey) -> Option<PlayerId> {
    let id = world.find_player(key)?;
    world.player(id)?.online.then_some(id)
}