//! A host and its guests, connected in memory instead of over a network.
//! They exchange the same bytes the shell sends.

use std::collections::BTreeMap;

use grovewright_core::guest::{Arrival, Guest};
use grovewright_core::host::{Host, Peer, MAX_ACTIONS_PER_TICK};
use grovewright_core::inventory::ITEM_STONE;
use grovewright_core::map::{Pos, Terrain};
use grovewright_core::mapgen::{outcrop_origin, TEST_MAP_SIZE};
use grovewright_core::player::{PlayerKey, MAX_KNOWN_PLAYERS};
use grovewright_core::protocol::{
    Action, Refusal, ToGuest, ToHost, CHECK_INTERVAL_TICKS, PROTOCOL_VERSION,
};
use grovewright_core::replay::Recording;
use grovewright_core::rng::Rng;
use grovewright_core::rules::rules_id;
use grovewright_core::world::{World, DIG_COOLDOWN_TICKS};

const HOST_KEY: PlayerKey = [100; 16];

/// A host and its guests in one process.
struct Lan {
    host: Host,
    guests: BTreeMap<Peer, Guest>,
    /// Guests whose connection the host closed, with the refusal they got.
    closed: BTreeMap<Peer, Option<Refusal>>,
    next_peer: Peer,
}

impl Lan {
    /// A host whose own player is in a new world from `seed`.
    fn new(seed: u64) -> Self {
        let mut host = Host::new(World::new(seed));
        host.join_local(HOST_KEY, "Host");
        host.step();
        Self {
            host,
            guests: BTreeMap::new(),
            closed: BTreeMap::new(),
            next_peer: 2,
        }
    }

    /// Connects a guest with a key made of `key` and lets it greet.
    fn connect(&mut self, key: u8, name: &str) -> Peer {
        let peer = self.next_peer;
        self.next_peer += 1;
        self.host.connect(peer);
        self.guests.insert(peer, Guest::new([key; 16], name));
        self.deliver();
        peer
    }

    /// Closes the connection of `peer`, as a guest that quits does.
    fn disconnect(&mut self, peer: Peer) {
        self.guests.remove(&peer);
        self.host.disconnect(peer);
    }

    /// Carries messages both ways until none are left.
    fn deliver(&mut self) {
        self.deliver_changed(|_, bytes| bytes);
    }

    /// Like [`Lan::deliver`], but `change` may alter what the host sends.
    fn deliver_changed(&mut self, mut change: impl FnMut(Peer, Vec<u8>) -> Vec<u8>) {
        loop {
            let mut moved = false;
            for (peer, guest) in self.guests.iter_mut() {
                for bytes in guest.take_outbox() {
                    self.host.receive(*peer, &bytes);
                    moved = true;
                }
            }
            for (peer, bytes) in self.host.take_outbox() {
                let bytes = change(peer, bytes);
                if let Some(guest) = self.guests.get_mut(&peer) {
                    let arrival = guest
                        .receive(&bytes)
                        .expect("the host's message is readable");
                    if let Arrival::Refused(refusal) = arrival {
                        self.closed.insert(peer, Some(refusal));
                    }
                }
                moved = true;
            }
            for peer in self.host.take_closing() {
                self.guests.remove(&peer);
                self.closed.entry(peer).or_insert(None);
                // The connection closes on both ends, and the host hears it.
                self.host.disconnect(peer);
            }
            if !moved {
                break;
            }
        }
    }

    /// Runs one tick on the host and lets every guest run what arrived.
    fn tick(&mut self) {
        self.host.step();
        self.deliver();
        for guest in self.guests.values_mut() {
            while guest.step().is_some() {}
        }
        self.deliver();
    }

    fn ticks(&mut self, count: u64) {
        for _ in 0..count {
            self.tick();
        }
    }

    fn guest(&mut self, peer: Peer) -> &mut Guest {
        self.guests.get_mut(&peer).expect("the guest is connected")
    }

    /// Asserts that every guest is in exactly the state of the host.
    fn assert_in_lockstep(&self) {
        let expected = self.host.world().state_hash();
        for (peer, guest) in &self.guests {
            assert_eq!(
                guest.world().map(World::state_hash),
                Some(expected),
                "guest {peer}"
            );
        }
    }

    /// The rock right beside the spawn clearing.
    fn rock(&self) -> Pos {
        outcrop_origin(self.host.world().map())
    }
}

/// The commands of the next tick as one guest receives them. Takes the
/// host's outbox, so the guests miss this tick.
fn commands_sent_to(host: &mut Host, peer: Peer) -> usize {
    host.take_outbox()
        .into_iter()
        .filter(|(to, _)| *to == peer)
        .find_map(|(_, bytes)| match ToGuest::decode(&bytes) {
            Ok(ToGuest::Tick { commands, .. }) => Some(commands.len()),
            _ => None,
        })
        .expect("a tick was sent")
}

#[test]
fn a_guest_gets_the_world_and_follows_every_tick() {
    let mut lan = Lan::new(5);
    let ada = lan.connect(1, "Ada");
    lan.ticks(3 * CHECK_INTERVAL_TICKS);

    lan.assert_in_lockstep();
    let guest = lan.guest(ada);
    assert!(!guest.has_diverged());
    let me = guest.local_player().expect("the guest is in the world");
    let world = guest.world().unwrap();
    assert_eq!(world.player(me).unwrap().name, "Ada");
    assert_eq!(
        world
            .players()
            .iter()
            .filter(|player| player.online)
            .count(),
        2
    );
}

#[test]
fn a_guest_joins_a_running_world_at_its_current_tick() {
    let mut lan = Lan::new(5);
    lan.ticks(100);
    let ada = lan.connect(1, "Ada");
    let tick = lan.host.world().tick();
    assert_eq!(lan.guest(ada).world().unwrap().tick(), tick);
    assert_eq!(
        lan.guest(ada).local_player(),
        None,
        "its join has not run yet"
    );

    lan.tick();
    assert!(lan.guest(ada).local_player().is_some());
    assert_eq!(lan.host.player_of(ada), lan.guest(ada).local_player());
    lan.assert_in_lockstep();
}

#[test]
fn the_actions_of_a_guest_change_the_world_of_everyone() {
    let mut lan = Lan::new(5);
    let ada = lan.connect(1, "Ada");
    let grace = lan.connect(2, "Grace");
    lan.tick();
    let rock = lan.rock();
    for _ in 0..3 {
        lan.guest(ada).act(Action::Dig { pos: rock });
        lan.ticks(DIG_COOLDOWN_TICKS + 1);
    }

    lan.assert_in_lockstep();
    let ada_player = lan.host.player_of(ada).unwrap();
    let grace_player = lan.host.player_of(grace).unwrap();
    let seen_by_grace = lan.guest(grace).world().unwrap().clone();
    assert_eq!(
        seen_by_grace.map().get(rock).unwrap().terrain,
        Terrain::Ground
    );
    assert_eq!(
        seen_by_grace
            .player(ada_player)
            .unwrap()
            .inventory
            .count(ITEM_STONE),
        1
    );
    assert_eq!(
        seen_by_grace
            .player(grace_player)
            .unwrap()
            .inventory
            .count(ITEM_STONE),
        0
    );
}

#[test]
fn the_host_takes_at_most_four_actions_per_player_and_tick() {
    let mut lan = Lan::new(5);
    let ada = lan.connect(1, "Ada");
    lan.tick();
    for x in 0..10 {
        lan.guest(ada).act(Action::Dig {
            pos: Pos::new(x, 1),
        });
        lan.host.act_local(Action::Dig {
            pos: Pos::new(x, 2),
        });
    }
    lan.deliver();
    lan.host.step();
    assert_eq!(
        commands_sent_to(&mut lan.host, ada),
        2 * MAX_ACTIONS_PER_TICK as usize
    );

    // The count starts again in the next tick.
    lan.host.act_local(Action::Dig {
        pos: Pos::new(1, 3),
    });
    lan.host.step();
    assert_eq!(commands_sent_to(&mut lan.host, ada), 1);
}

#[test]
fn a_guest_of_another_protocol_version_is_told_so() {
    let mut lan = Lan::new(5);
    let peer = 9;
    lan.host.connect(peer);
    let foreign = ToHost::ForeignHello {
        protocol: PROTOCOL_VERSION + 1,
    };
    lan.host.receive(peer, &foreign.encode());
    assert_eq!(
        lan.host.take_outbox(),
        vec![(peer, ToGuest::Refused(Refusal::OtherProtocol).encode())]
    );
    assert_eq!(lan.host.take_closing(), vec![peer]);
}

#[test]
fn a_guest_with_other_rules_is_refused() {
    let mut lan = Lan::new(5);
    let peer = 9;
    lan.host.connect(peer);
    let hello = ToHost::Hello {
        rules: rules_id() ^ 1,
        key: [1; 16],
        name: "Ada".to_owned(),
    };
    lan.host.receive(peer, &hello.encode());
    assert_eq!(
        lan.host.take_outbox(),
        vec![(peer, ToGuest::Refused(Refusal::OtherRules).encode())]
    );
    assert_eq!(lan.host.take_closing(), vec![peer]);
}

#[test]
fn a_fifth_player_is_refused_even_before_the_others_have_joined() {
    let mut lan = Lan::new(5);
    let guests = [1, 2, 3].map(|key| lan.connect(key, "Guest"));
    let fifth = lan.connect(4, "Fifth");
    assert_eq!(lan.closed.get(&fifth), Some(&Some(Refusal::WorldFull)));

    lan.tick();
    let late = lan.connect(5, "Late");
    assert_eq!(lan.closed.get(&late), Some(&Some(Refusal::WorldFull)));
    assert_eq!(lan.host.guest_count(), 3);

    // Once a guest has left, there is room again.
    lan.disconnect(guests[0]);
    lan.tick();
    let next = lan.connect(6, "Next");
    lan.tick();
    assert!(lan.host.player_of(next).is_some());
    lan.assert_in_lockstep();
}

#[test]
fn a_world_that_knows_sixteen_players_refuses_a_new_one_but_not_an_old_one() {
    let mut lan = Lan::new(5);
    // The host and fifteen guests who come and go.
    for key in 1..MAX_KNOWN_PLAYERS as u8 {
        let peer = lan.connect(key, "Visitor");
        lan.tick();
        lan.disconnect(peer);
        lan.tick();
    }
    assert_eq!(lan.host.world().players().len(), MAX_KNOWN_PLAYERS);

    let stranger = lan.connect(200, "Stranger");
    assert_eq!(
        lan.closed.get(&stranger),
        Some(&Some(Refusal::TooManyPlayers))
    );
    let regular = lan.connect(3, "Visitor");
    lan.tick();
    assert!(lan.host.player_of(regular).is_some());
}

#[test]
fn a_key_that_is_in_the_world_already_is_refused() {
    let mut lan = Lan::new(5);
    let ada = lan.connect(1, "Ada");
    let twin = lan.connect(1, "Ada");
    assert_eq!(lan.closed.get(&twin), Some(&Some(Refusal::AlreadyPlaying)));
    lan.tick();
    let impostor = lan.connect(HOST_KEY[0], "Host");
    assert_eq!(
        lan.closed.get(&impostor),
        Some(&Some(Refusal::AlreadyPlaying))
    );
    assert!(lan.host.player_of(ada).is_some());
}

#[test]
fn a_guest_that_leaves_is_offline_everywhere_and_keeps_its_bag() {
    let mut lan = Lan::new(5);
    let ada = lan.connect(1, "Ada");
    let grace = lan.connect(2, "Grace");
    lan.tick();
    let rock = lan.rock();
    for _ in 0..3 {
        lan.guest(ada).act(Action::Dig { pos: rock });
        lan.ticks(DIG_COOLDOWN_TICKS + 1);
    }
    let player = lan.host.player_of(ada).unwrap();

    lan.disconnect(ada);
    lan.tick();
    assert!(!lan.host.world().player(player).unwrap().online);
    assert!(
        !lan.guest(grace)
            .world()
            .unwrap()
            .player(player)
            .unwrap()
            .online
    );
    lan.assert_in_lockstep();

    let back = lan.connect(1, "Ada");
    lan.tick();
    assert_eq!(lan.host.player_of(back), Some(player));
    assert_eq!(lan.guest(back).local_player(), Some(player));
    assert_eq!(
        lan.host
            .world()
            .player(player)
            .unwrap()
            .inventory
            .count(ITEM_STONE),
        1
    );
    lan.assert_in_lockstep();
}

#[test]
fn a_guest_that_goes_before_its_join_ran_never_enters() {
    let mut lan = Lan::new(5);
    let ada = lan.connect(1, "Ada");
    lan.disconnect(ada);
    lan.tick();
    assert_eq!(lan.host.world().find_player(&[1; 16]), None);
    assert_eq!(lan.host.guest_count(), 0);
}

#[test]
fn a_guest_that_breaks_the_protocol_is_closed() {
    let mut lan = Lan::new(5);
    let act = ToHost::Act(Action::Dig {
        pos: Pos::new(1, 1),
    })
    .encode();
    let broken = [vec![], vec![0xEE], vec![2; 200], act.clone()];
    for (index, bytes) in broken.iter().enumerate() {
        let peer = 50 + index as Peer;
        lan.host.connect(peer);
        lan.host.receive(peer, bytes);
        assert_eq!(lan.host.take_closing(), vec![peer], "message {index}");
    }

    // Messages from a connection the host does not know are ignored.
    lan.host.receive(77, &act);
    assert!(lan.host.take_outbox().is_empty());
    assert!(lan.host.take_closing().is_empty());

    // A second greeting from a guest in the world closes it, and its
    // player leaves.
    let ada = lan.connect(1, "Ada");
    lan.tick();
    let player = lan.host.player_of(ada).unwrap();
    let hello = ToHost::Hello {
        rules: rules_id(),
        key: [1; 16],
        name: "Ada".to_owned(),
    };
    lan.host.receive(ada, &hello.encode());
    assert_eq!(lan.host.take_closing(), vec![ada]);
    lan.host.step();
    assert!(!lan.host.world().player(player).unwrap().online);
}

#[test]
fn a_guest_that_diverges_gets_the_world_again_and_keeps_the_evidence() {
    let mut lan = Lan::new(5);
    let ada = lan.connect(1, "Ada");
    lan.tick();
    // The host's player hits the rock, but Ada's copy of that tick loses the
    // command, as if her machine had run it differently.
    lan.host.act_local(Action::Dig { pos: lan.rock() });
    lan.host.step();
    lan.deliver_changed(|_, bytes| match ToGuest::decode(&bytes) {
        Ok(ToGuest::Tick { tick, check, .. }) => ToGuest::Tick {
            tick,
            commands: vec![],
            check,
        }
        .encode(),
        _ => bytes,
    });
    while lan.guest(ada).step().is_some() {}
    assert!(
        !lan.guest(ada).has_diverged(),
        "nobody notices before the next checksum"
    );

    lan.ticks(CHECK_INTERVAL_TICKS);
    lan.assert_in_lockstep();
    assert!(!lan.guest(ada).has_diverged());
    let divergences = lan.host.take_divergences();
    assert_eq!(divergences.len(), 1);
    assert_eq!(divergences[0].peer, ada);

    let evidence = lan
        .guest(ada)
        .take_diverged_recording()
        .expect("the recording of the diverged world is kept");
    let replayed = Recording::verify(&evidence).expect("the evidence replays");
    assert_eq!(replayed.state_hash(), divergences[0].hash);
    assert_eq!(lan.guest(ada).take_diverged_recording(), None);
}

#[test]
fn a_long_busy_session_stays_in_lockstep() {
    let mut lan = Lan::new(77);
    let mut rng = Rng::new(9);
    let mut peers: Vec<Peer> = [1, 2, 3].map(|key| lan.connect(key, "Busy")).to_vec();
    let area = TEST_MAP_SIZE / 3;
    let random_pos = |rng: &mut Rng| {
        Pos::new(
            (area + rng.below(area)) as i32,
            (area + rng.below(area)) as i32,
        )
    };
    for tick in 0..2_000 {
        for &peer in &peers {
            if rng.below(3) == 0 {
                let pos = random_pos(&mut rng);
                lan.guest(peer).act(Action::Dig { pos });
            }
        }
        if rng.below(2) == 0 {
            let pos = random_pos(&mut rng);
            lan.host.act_local(Action::Dig { pos });
        }
        if tick == 700 {
            lan.disconnect(peers.remove(1));
        }
        if tick == 1_100 {
            peers.push(lan.connect(2, "Busy"));
        }
        lan.tick();
    }

    lan.assert_in_lockstep();
    assert!(lan.host.take_divergences().is_empty());
    for guest in lan.guests.values() {
        let replayed = guest.recording().unwrap().replay().unwrap();
        assert_eq!(&replayed, guest.world().unwrap());
    }
}