use serde::de::DeserializeOwned;
use std::collections::{HashMap, VecDeque};
use std::{mem, ptr, sync::Arc, time::SystemTime};
use ustr::{Ustr, UstrMap, UstrSet};

use crate::{
    ArgumentError, AsLocationId, ConnectionOptions, Error, Event, Game, ItemHandling, Iter,
    LocatedItem, Location, Player, Print, ProtocolError, ReceivedItem, SignedDuration, Socket,
    UnsizedIter, UpdatedField, Version, protocol::*,
};

mod bounce_options;
mod create_hints_options;
mod death_link_options;

pub use bounce_options::*;
pub use create_hints_options::*;
pub use death_link_options::*;

/// The version of the Archipelago server that this client supports.
const VERSION: NetworkVersion = NetworkVersion {
    major: 0,
    minor: 6,
    build: 0,
    class: String::new(),
};

/// The client that talks to the Archipelago server using the Archipelago
/// protocol.
///
/// The generic type `S` is used to deserialize the slot data in the initial
/// `Connected` message. By default, it will decode the slot data as a dynamic
/// JSON blob. If `S = ()`, this will not request slot data from the server at
/// all.
pub struct Client<S: DeserializeOwned + 'static = serde_json::Value> {
    socket: Socket<S>,

    // == Session information
    game: *const Game,
    server_version: Version,
    generator_version: Version,
    server_tags: UstrSet,
    password_required: bool,
    permissions: PermissionMap,
    hint_cost_percentage: u8,
    hint_points_per_check: u64,
    hint_points: u64,
    seed_name: String,
    games: UstrMap<Game>,
    slot_data: S,

    /// The difference between the server's notion of the current time and ours.
    /// We use this to normalize timestamps, under the assumption that they
    /// should match the server's time.
    server_skew: SignedDuration,

    /// A map from `(team, slot)` to the player with that team and slot.
    players: HashMap<(u32, u32), Arc<Player>>,

    /// The key for the current player in [players].
    player_key: (u32, u32),

    /// A map from location IDs for this game to booleans indicating whether or
    /// not they've been checked.
    local_locations_checked: HashMap<i64, bool>,

    /// A list of all the items this client has ever received. This is
    /// overwritten if the server sends a new [ServerMessage::ReceivedItems]
    /// with index 0.
    received_items: Vec<ReceivedItem>,

    /// Senders for [Client.scout_locations].
    location_scout_senders: VecDeque<oneshot::Sender<Result<Vec<LocatedItem>, Error>>>,

    /// Senders for [Client.get].
    get_senders: VecDeque<oneshot::Sender<Result<HashMap<String, serde_json::Value>, Error>>>,
}

impl<S: DeserializeOwned + 'static> Client<S> {
    /// Asynchronously initializes a client connection to an Archipelago server.
    ///
    /// If the `url` doesn't have a protocol provided, this tries `wss://`
    /// followed by `ws://`. If it doesn't have a port, it defaults to the
    /// Archipelago default port 38281.
    ///
    /// If `game` is `None`, [ConnectionOptions::tags] must contain at least one
    /// of [tags::HINT_GAME], [tags::TRACKER] or [tags::TEXT_ONLY]. In this
    /// case, the server won't validate that the slot you connect to is playing
    /// a particular game.
    ///
    /// [tags::HINT_GAME]: crate::tags::HINT_GAME
    /// [tags::TRACKER]: crate::tags::TRACKER
    /// [tags::TEXT_ONLY]: crate::tags::TEXT_ONLY
    ///
    /// See [ConnectionOptions] for details about optional arguments and their
    /// defaults.
    pub async fn connect(
        url: impl Into<String>,
        name: impl Into<Ustr>,
        game: Option<impl Into<Ustr>>,
        options: ConnectionOptions,
    ) -> Result<Client<S>, Error> {
        if game.is_none()
            && !options.tags.contains(&"HintGame".into())
            && !options.tags.contains(&"Tracker".into())
            && !options.tags.contains(&"TextOnly".into())
        {
            return Err(ArgumentError::MissingGame { tags: options.tags }.into());
        }

        let url = url.into();
        let mut socket = if url.as_str().starts_with("ws://") || url.as_str().starts_with("wss://")
        {
            Socket::connect(
                url,
                #[cfg(feature = "rustls")]
                options.rustls_config,
            )
            .await?
        } else {
            match Socket::connect(
                format!("wss://{}", url),
                #[cfg(feature = "rustls")]
                options.rustls_config.clone(),
            )
            .await
            {
                Ok(socket) => socket,
                Err(Error::WebSocket(err)) => {
                    match Socket::connect(
                        format!("ws://{}", url),
                        #[cfg(feature = "rustls")]
                        options.rustls_config,
                    )
                    .await
                    {
                        Ok(socket) => socket,
                        Err(_) => return Err(err.into()),
                    }
                }
                Err(err) => return Err(err),
            }
        };

        log::debug!("Awaiting RoomInfo...");
        let room_info = match socket.recv_async().await? {
            ServerMessage::RoomInfo(room_info) => room_info,
            message => return Err(Self::unexpected_response(message, "RoomInfo")),
        };

        log::debug!("Loading Cached DataPackages...");
        let cache = options.cache.unwrap_or_default();
        let mut packages = cache
            .load_data_packages(&room_info.datapackage_checksums)
            .await;
        // Determine which games we are missing by comparing the checksums
        // received with what we found in the cache.
        let missing = room_info
            .datapackage_checksums
            .keys()
            .filter(|k| !packages.contains_key(*k))
            .map(|k| k.to_string())
            .collect::<Vec<_>>();
        if !missing.is_empty() {
            log::debug!("Awaiting DataPackage...");
            socket.send(ClientMessage::GetDataPackage(GetDataPackage {
                games: Some(missing),
            }))?;
            let received_dp = match socket.recv_async().await? {
                ServerMessage::DataPackage(DataPackage { data }) => data,
                message => return Err(Self::unexpected_response(message, "DataPackage")),
            };
            log::debug!("Writing new entries to cache...");
            cache.store_data_packages(&received_dp.games).await;
            packages.extend(received_dp.games);
        }
        let data_package = DataPackageObject { games: packages };

        log::debug!("Awaiting Connected...");
        let mut version = VERSION.clone();
        version.class = "Version".into();
        socket.send(ClientMessage::Connect(Connect {
            password: options.password,
            game: game.map(|g| g.into()),
            name: name.into(),
            // Specify something useful here if
            // ArchipelagoMW/Archipelago#998 ever gets sorted out.
            uuid: "".into(),
            version: version.clone(),
            items_handling: options.item_handling.into(),
            tags: options.tags,
            slot_data: !try_specialize::static_type_eq::<S, ()>(),
        }))?;

        let connected = match socket.recv_async().await? {
            ServerMessage::Connected(connected) => connected,
            ServerMessage::ConnectionRefused(ConnectionRefused { errors }) => {
                return Err(Error::ConnectionRefused(
                    errors.into_iter().map(|e| e.into()).collect(),
                ));
            }
            message => return Err(Self::unexpected_response(message, "Connected")),
        };

        let client = Client::new(socket, room_info, data_package, connected)?;
        log::info!("Archipelago connection initialized successfully");
        Ok(client)
    }

    /// Creates a new client with all available initial information.
    fn new(
        socket: Socket<S>,
        room_info: RoomInfo,
        data_package: DataPackageObject,
        connected: Connected<S>,
    ) -> Result<Self, Error> {
        let server_skew = SignedDuration::difference(SystemTime::now(), room_info.time);
        let total_locations = connected.checked_locations.len() + connected.missing_locations.len();

        let teams = connected
            .players
            .iter()
            .map(|p| p.team)
            .max()
            .ok_or(ProtocolError::EmptyPlayers)?
            + 1;

        let mut players = HashMap::<(u32, u32), Arc<Player>>::new();
        for player in connected.players {
            let slot_info = connected
                .slot_info
                .get(&player.slot)
                .ok_or(ProtocolError::MissingSlotInfo(player.slot))?;
            // Groups always come after players in the slot list and can't
            // contain other groups, so only considering the players we've
            // already hydrated should be safe.
            let player = Player::hydrate(player, slot_info, &players)?;
            players.insert((player.team(), player.slot()), player.into());
        }
        for team in 0..teams {
            players.insert((team, 0), Player::archipelago(team).into());
        }

        let player_key = (connected.team, connected.slot);
        if !players.contains_key(&player_key) {
            return Err(ProtocolError::MissingPlayer {
                team: connected.team,
                slot: connected.slot,
            }
            .into());
        }

        let mut games = data_package
            .games
            .into_iter()
            .map(|(name, data)| (name, Game::hydrate(name, data)))
            .collect::<UstrMap<_>>();
        for game_name in room_info.games {
            games
                .entry(game_name)
                .or_insert_with(|| Game::no_data_package(game_name));
        }

        let game = connected
            .slot_info
            .get(&connected.slot)
            .ok_or(ProtocolError::MissingSlotInfo(connected.slot))?
            .game;
        let game = games
            .get(&game)
            .ok_or(ProtocolError::MissingGameData(game))?;
        let game_ptr = ptr::from_ref(game);

        let mut local_locations_checked = HashMap::with_capacity(total_locations);
        for id in connected.missing_locations {
            game.verify_location(id)?;
            local_locations_checked.insert(id, false);
        }
        for id in connected.checked_locations {
            game.verify_location(id)?;
            local_locations_checked.insert(id, true);
        }

        Ok(Client {
            socket,
            game: game_ptr,
            server_version: room_info.version.into(),
            generator_version: room_info.generator_version.into(),
            server_tags: room_info.tags,
            password_required: room_info.password_required,
            permissions: room_info.permissions,
            hint_cost_percentage: room_info.hint_cost,
            hint_points_per_check: room_info.location_check_points,
            hint_points: connected.hint_points,
            seed_name: room_info.seed_name,
            games,
            slot_data: connected.slot_data,
            server_skew,
            players,
            player_key,
            local_locations_checked,
            received_items: Default::default(),
            location_scout_senders: Default::default(),
            get_senders: Default::default(),
        })
    }

    /// Returns an error indicating that [message] is unexpected.
    fn unexpected_response(message: ServerMessage<S>, expected: &'static str) -> Error {
        if let ServerMessage::InvalidPacket(invalid) = message {
            Error::InvalidPacket(invalid.text)
        } else {
            ProtocolError::UnexpectedResponse {
                actual: message.type_name(),
                expected,
            }
            .into()
        }
    }

    // == Session information

    /// The game that's currently being played.
    pub fn this_game(&self) -> &Game {
        // Safety: This game is stored in [games], which we own and which is
        // never mutated.
        unsafe { &*self.game }
    }

    /// The version of Archipelago which the server is running.
    pub fn server_version(&self) -> &Version {
        &self.server_version
    }

    /// The version of Archipelago that generated the multiworld.
    pub fn generator_version(&self) -> &Version {
        &self.generator_version
    }

    /// The server's special features or capabilities.
    pub fn server_tags(&self) -> &UstrSet {
        &self.server_tags
    }

    /// Whether this Archipelago multiworld requires a password to join.
    pub fn password_required(&self) -> bool {
        self.password_required
    }

    /// The permissions for distributing all items after a player reaches their
    /// goal to other players awaiting them.
    pub fn release_permission(&self) -> Permission {
        self.permissions.release
    }

    /// The permissions for collecting all items after a player reaches their
    /// goal.
    pub fn collect_permission(&self) -> Permission {
        self.permissions.collect
    }

    /// The permissions for a player querying the items remaining in their run.
    pub fn remaining_permission(&self) -> Permission {
        self.permissions.remaining
    }

    /// The number of hint points the player must accumulate in order to access
    /// a single hint.
    pub fn points_per_hint(&self) -> u64 {
        (self.local_locations_checked.len() as u64) * u64::from(self.hint_cost_percentage) / 100
    }

    /// The number of hint points granted for each location a player checks.
    pub fn hint_points_per_check(&self) -> u64 {
        self.hint_points_per_check
    }

    /// The number of hint points the player currently has.
    pub fn hint_points(&self) -> u64 {
        self.hint_points
    }

    /// The uniquely-identifying name of the generated multiworld.
    ///
    /// If the same multiworld is hosted in multiple rooms, this will be the
    /// same across those rooms.
    pub fn seed_name(&self) -> &str {
        self.seed_name.as_str()
    }

    /// A map from the names of each game in this multiworld to metadata about
    /// those games.
    pub fn games(&self) -> impl Iter<&Game> {
        self.games.values()
    }

    /// Returns the game with the given `name`, if one is in this multiworld.
    ///
    /// Unlike [games](Self::games), this will return the special
    /// [Game::archipelago] game.
    pub fn game(&self, name: impl Into<Ustr>) -> Option<&Game> {
        let name = name.into();
        // Safety: We own the name for the duration of the call.
        self.games.get(&name).or_else(|| {
            let archipelago = Game::archipelago();
            if name == archipelago.name() {
                Some(archipelago)
            } else {
                None
            }
        })
    }

    /// Returns the game in this multiworld with the given `name`. Panics if
    /// there's no game with that name.
    pub fn assert_game(&self, name: impl Into<Ustr>) -> &Game {
        let name = name.into();
        self.game(name)
            .unwrap_or_else(|| panic!("multiworld doesn't contain a game named \"{}\"", name))
    }

    /// Returns the game in this multiworld with the given [name]. Returns an
    /// error if there's no game with that name.
    pub(crate) fn game_or_err(&self, name: impl Into<Ustr>) -> Result<&Game, Error> {
        let name = name.into();
        self.game(name)
            .ok_or_else(|| ProtocolError::MissingGameData(name).into())
    }

    /// The player that's currently connected to the multiworld.
    pub fn this_player(&self) -> &Player {
        self.players[&self.player_key].as_ref()
    }

    /// All players in the multiworld.
    pub fn players(&self) -> impl Iter<&Player> {
        self.players.values().map(|p| p.as_ref())
    }

    /// The player on the given `team` playing the given `slot`, if one exists.
    ///
    /// See also [teammate](Self::teammate) to only check the current player's team.
    pub fn player(&self, team: u32, slot: u32) -> Option<&Player> {
        self.players.get(&(team, slot)).map(|p| p.as_ref())
    }

    /// Clones the Arc for the player on the given `team` playing the given
    /// `slot`.
    pub(crate) fn player_arc(&self, team: u32, slot: u32) -> Result<Arc<Player>, Error> {
        self.players
            .get(&(team, slot))
            .cloned()
            .ok_or_else(|| ProtocolError::MissingPlayer { team, slot }.into())
    }

    /// The player on the given `team` playing the given `slot`. Panics if this
    /// player doesn't exist.
    ///
    /// See also [assert_teammate](Self::assert_teammate) to only check the
    /// current player's team.
    pub fn assert_player(&self, team: u32, slot: u32) -> &Player {
        self.player(team, slot).unwrap_or_else(|| {
            if self.players.keys().any(|k| k.0 == team) {
                panic!("no player with slot {}", slot);
            } else {
                panic!("no team with ID {}", team);
            }
        })
    }

    /// The player playing the given `slot` on the current player's team, if one
    /// exists.
    pub fn teammate(&self, slot: u32) -> Option<&Player> {
        self.player(self.player_key.0, slot)
    }

    /// A clone of the [Arc] for the player playing the given [slot] on the
    /// current player's team.
    pub(crate) fn teammate_arc(&self, slot: u32) -> Result<Arc<Player>, Error> {
        self.player_arc(self.player_key.0, slot)
    }

    /// The player playing the given `slot` on the current player's team. Panics
    /// if this player doesn't exist.
    pub fn assert_teammate(&self, slot: u32) -> &Player {
        self.assert_player(self.player_key.0, slot)
    }

    /// Returns whether the local location with the given ID has been checked
    /// (either by us or by other players doing co-op).
    ///
    /// This may only be called for locations in the connected game's world.
    /// Panics if it's called with a location ID that doesn't exist for this
    /// world.
    pub fn is_local_location_checked(&self, id: impl AsLocationId) -> bool {
        let id = id.as_location_id();
        *self.local_locations_checked.get(&id).unwrap_or_else(|| {
            panic!(
                "Archipelago location ID {} doesn't exist for {}",
                id,
                self.this_game().name()
            )
        })
    }

    /// Returns all the locations that the player has already checked.
    pub fn checked_locations(&self) -> impl UnsizedIter<Location> {
        let game = self.this_game();
        self.local_locations_checked
            .iter()
            .filter(|(_, checked)| **checked)
            .map(|(id, _)| game.assert_location(*id))
    }

    /// Returns all the locations that the player has not yet checked.
    pub fn unchecked_locations(&self) -> impl UnsizedIter<Location> {
        let game = self.this_game();
        self.local_locations_checked
            .iter()
            .filter(|(_, checked)| !**checked)
            .map(|(id, _)| game.assert_location(*id))
    }

    /// Returns all items that have ever been received by this client. Note that
    /// this will be empty until the initial [Event::ReceivedItems] is emitted.
    pub fn received_items(&self) -> &[ReceivedItem] {
        &self.received_items
    }

    /// Returns the slot data provided by the apworld.
    pub fn slot_data(&self) -> &S {
        &self.slot_data
    }

    // == Requests

    /// Updates the current connection settings with new `item_handling` and/or
    /// `tags`.
    pub fn update_connection(
        &mut self,
        item_handling: Option<ItemHandling>,
        tags: Option<impl IntoIterator<Item: Into<Ustr>>>,
    ) -> Result<(), Error> {
        self.socket
            .send(ClientMessage::ConnectUpdate(ConnectUpdate {
                items_handling: item_handling.map(|i| i.into()),
                tags: tags.map(|ts| ts.into_iter().map(|t| t.into()).collect()),
            }))
    }

    /// Requests that the server resends all items this client has ever
    /// received. This will appear as an [Event::ReceivedItems] with index 0.
    pub fn sync(&mut self) -> Result<(), Error> {
        self.socket.send(ClientMessage::Sync)
    }

    /// Notifies the server that the given `locations` have been checked.
    pub fn mark_checked(
        &mut self,
        locations: impl IntoIterator<Item = impl AsLocationId>,
    ) -> Result<(), Error> {
        let locations = self.verify_local_locations(locations)?;
        self.socket
            .send(ClientMessage::LocationChecks(LocationChecks {
                locations: locations.clone(),
            }))?;

        for id in locations {
            if matches!(self.local_locations_checked.insert(id, true), Some(false)) {
                self.hint_points += self.hint_points_per_check;
            }
        }
        Ok(())
    }

    /// Sends a request to the server that can serve one or both of two
    /// purposes:
    ///
    /// * Informing the client which items exist at which location. This
    ///   information will be made available by the returned
    ///   [oneshot::Receiver].
    ///
    /// * Informing the server of locations that the client has seen but not
    ///   checked. If [CreateAsHint.All] or [CreateAsHint.New] is passed,
    ///   scouted locations will be broadcast as hints without deducting hint
    ///   points from the player.
    pub fn scout_locations(
        &mut self,
        locations: impl IntoIterator<Item = impl AsLocationId>,
        create_as_hint: CreateAsHint,
    ) -> oneshot::Receiver<Result<Vec<LocatedItem>, Error>> {
        let (sender, receiver) = oneshot::channel();
        match self
            .verify_local_locations(locations)
            .and_then(|locations| {
                self.socket
                    .send(ClientMessage::LocationScouts(LocationScouts {
                        locations,
                        create_as_hint,
                    }))
            }) {
            Ok(()) => self.location_scout_senders.push_back(sender),
            // If `send()` returns an error, that means that the receiver was
            // dropped, which is fine to silently ignore.
            Err(err) => mem::drop(sender.send(Err(err))),
        }
        receiver
    }

    /// Create hints for the specified locations on the server. Locations that
    /// already have hints will be ignored.
    ///
    /// This is a shorthand for calling
    /// [create_hints_with_options](Self::create_hints_with_options) with only
    /// default options. It create hints for the current player's slot at
    /// [HintStatus::Unspecified].
    pub fn create_hints(
        &mut self,
        locations: impl IntoIterator<Item = impl AsLocationId>,
    ) -> Result<(), Error> {
        self.create_hints_with_options(locations, Default::default())
    }

    /// Create hints for the specified locations on the server. Locations that
    /// already have hints will be ignored.
    pub fn create_hints_with_options(
        &mut self,
        locations: impl IntoIterator<Item = impl AsLocationId>,
        options: CreateHintsOptions,
    ) -> Result<(), Error> {
        let slot = options.slot.unwrap_or(self.player_key.1);
        self.verify_game_locations(
            self.assert_game(self.verify_teammate(slot)?.game()),
            locations,
        )
        .and_then(|locations| {
            self.socket.send(ClientMessage::CreateHints(CreateHints {
                locations,
                player: slot,
                status: options.status,
            }))
        })
    }

    /// Updates the status of the given hint on the server.
    ///
    /// This allows the player to indicate which hinted items from other worlds
    /// they care about.
    pub fn update_hint(
        &mut self,
        slot: u32,
        location: impl AsLocationId,
        status: HintStatus,
    ) -> Result<(), Error> {
        let location = location.as_location_id();
        let player = self.verify_teammate(slot)?;
        let game = self.assert_game(player.game());
        if !game.has_location(location) {
            return Err(ArgumentError::InvalidLocation {
                location,
                game: game.name(),
            }
            .into());
        }

        self.socket.send(ClientMessage::UpdateHint(UpdateHint {
            player: player.slot(),
            location,
            status,
        }))
    }

    /// Notifies the server that the client has the given `status`.
    pub fn set_status(&mut self, status: ClientStatus) -> Result<(), Error> {
        self.socket
            .send(ClientMessage::StatusUpdate(StatusUpdate { status }))
    }

    /// Broadcasts `text` to all teammates in the multiworld.
    pub fn say(&mut self, text: String) -> Result<(), Error> {
        self.socket.send(ClientMessage::Say(Say { text }))
    }

    /// Broadcasts `data` to other clients in the multiworld.
    pub fn bounce(&mut self, data: serde_json::Value, options: BounceOptions) -> Result<(), Error> {
        self.socket.send(ClientMessage::Bounce(Bounce {
            games: options.games,
            slots: options.slots,
            tags: options.tags,
            data: BounceData::Generic(Some(data)),
        }))
    }

    /// Notifies the player's teammates with death link enabled that the player
    /// has died.
    pub fn death_link(&mut self, options: DeathLinkOptions) -> Result<(), Error> {
        let mut tags = options.tags.unwrap_or_default();
        tags.insert(*DEATH_LINK_TAG);
        self.socket.send(ClientMessage::Bounce(Bounce {
            games: options.games,
            slots: options.slots,
            tags: Some(tags),
            data: BounceData::DeathLink(DeathLink {
                // Subtract the server delay so that we're sending our best
                // guess of the time on the server when the death occurred.
                time: options.time.unwrap_or(SystemTime::now()) - self.server_skew,
                cause: options.cause,
                source: options
                    .source
                    .unwrap_or_else(|| self.this_player().alias().to_string()),
            }),
        }))
    }

    /// Retrieves custom data from the server's data store. The specific
    /// structure of the data is up to the clients that set it.
    ///
    /// This can also retrieve special fields generated by the Archipelago
    /// server. See [the protocol documentation] for details.
    ///
    /// [the protocol documentation]: https://github.com/ArchipelagoMW/Archipelago/blob/main/docs/network%20protocol.md#get
    pub fn get(
        &mut self,
        keys: impl IntoIterator<Item = impl Into<String>>,
    ) -> oneshot::Receiver<Result<HashMap<String, serde_json::Value>, Error>> {
        let (sender, receiver) = oneshot::channel();
        match self.socket.send(ClientMessage::Get(Get {
            keys: keys.into_iter().map(|k| k.into()).collect(),
        })) {
            Ok(()) => self.get_senders.push_back(sender),
            Err(err) => mem::drop(sender.send(Err(err))),
        }
        receiver
    }

    /// Sets custom data in the server's data store. The specific structure of
    /// the data is up to the clients that set it.
    ///
    /// If `emit_event` is `true`, [update](Self::update) will eventually emit
    /// [Event::KeyChanged] for this key, even if it's not otherwise being
    /// watched.
    pub fn set(
        &mut self,
        key: impl Into<String>,
        value: serde_json::Value,
        emit_event: bool,
    ) -> Result<(), Error> {
        self.socket.send(ClientMessage::Set(Set {
            key: key.into(),
            default: serde_json::Value::Null,
            operations: vec![DataStorageOperation::Replace(value)],
            want_reply: emit_event,
        }))
    }

    /// Changes custom data in the server's data store. The specific structure
    /// of the data is up to the clients that set it.
    ///
    /// This applies each change in `operations` in order. If the key doesn't
    /// already have a value, `default` is used.
    ///
    /// If `emit_event` is `true`, [update](Self::update) will eventually emit
    /// [Event::KeyChanged] for this key, even if it's not otherwise being
    /// watched.
    pub fn change(
        &mut self,
        key: impl Into<String>,
        default: serde_json::Value,
        operations: impl IntoIterator<Item = DataStorageOperation>,
        emit_event: bool,
    ) -> Result<(), Error> {
        self.socket.send(ClientMessage::Set(Set {
            key: key.into(),
            default,
            operations: operations.into_iter().collect(),
            want_reply: emit_event,
        }))
    }

    /// Watches the given `keys` in the server's data store. Any time the key is
    /// set (even if it doesn't change), [Event::KeyChanged] will be emitted.
    pub fn watch(
        &mut self,
        keys: impl IntoIterator<Item = impl Into<String>>,
    ) -> Result<(), Error> {
        self.socket.send(ClientMessage::SetNotify(SetNotify {
            keys: keys.into_iter().map(|k| k.into()).collect(),
        }))
    }

    /// Converts [locations] to a vector and verifies that they're valid for the
    /// current game.
    fn verify_local_locations(
        &self,
        locations: impl IntoIterator<Item = impl AsLocationId>,
    ) -> Result<Vec<i64>, Error> {
        self.verify_game_locations(self.this_game(), locations)
    }

    /// Converts [locations] to a vector and verifies that they're valid for the
    /// specified game.
    fn verify_game_locations(
        &self,
        game: &Game,
        locations: impl IntoIterator<Item = impl AsLocationId>,
    ) -> Result<Vec<i64>, Error> {
        locations
            .into_iter()
            .map(|l| {
                let id = l.as_location_id();
                if game.has_location(id) {
                    Ok(id)
                } else {
                    Err(ArgumentError::InvalidLocation {
                        location: id,
                        game: game.name(),
                    }
                    .into())
                }
            })
            .collect()
    }

    /// Returns the [Player] for [slot] on the current team and verifies that
    /// it's a valid slot number.
    fn verify_teammate(&self, slot: u32) -> Result<&Player, Error> {
        self.teammate(slot)
            .ok_or(ArgumentError::InvalidSlot(slot).into())
    }

    // == Event handling

    /// Returns any pending [Event]s from the Archipelago server and updates the
    /// rest of the client's state to accommodate new information.
    ///
    /// This call never blocks, and is expected to be called repeatedly in order
    /// to check for new messages from the underlying connection to Archipelago.
    /// Typically a caller that's integrated Archipelago into a game loop will
    /// call this once each frame.
    ///
    /// Most errors are fatal, and if emitted mean that the client will not emit
    /// any more events and should be discarded and reconnected. Some
    /// (specifically [Error::ProtocolError]s) are recoverable, though, and the
    /// client will continue to emit additional events after they're emitted if
    /// it's not dropped. You can detect which errors are fatal using
    /// [Error.is_fatal].
    pub fn update(&mut self) -> Vec<Event> {
        self.socket
            .recv_all()
            .into_iter()
            .filter_map(|message| match message {
                Ok(message) => self.handle_message(message),
                Err(err) => Some(Event::Error(err)),
            })
            .collect()
    }

    /// Returns a single pending [Event] from the Archipelago server and updates
    /// the rest of the client's state to accommodate new information.
    ///
    /// This call never blocks, and is expected to be called repeatedly in order
    /// to check for new messages from the underlying connection to Archipelago.
    /// Typically a caller that's integrated Archipelago into a game loop will
    /// call this repeatedly each frame, or call [Self::update] once.
    ///
    /// Most errors are fatal, and if emitted mean that the client will not emit
    /// any more events and should be discarded and reconnected. Some
    /// (specifically [Error::ProtocolError]s) are recoverable, though, and the
    /// client will continue to emit additional events after they're emitted if
    /// it's not dropped. You can detect which errors are fatal using
    /// [Error.is_fatal].
    pub fn try_next_event(&mut self) -> Option<Event> {
        self.socket.try_recv().and_then(|message| match message {
            Ok(message) => self.handle_message(message),
            Err(err) => Some(Event::Error(err)),
        })
    }

    /// Handles a single message, converting it into an event for the user if
    /// necessary.
    fn handle_message(&mut self, message: ServerMessage<S>) -> Option<Event> {
        match message {
            ServerMessage::RawPrint(print) => Some(match Print::hydrate(print, self) {
                Ok(print) => Event::Print(print),
                Err(err) => Event::Error(err),
            }),

            ServerMessage::PlainPrint(PlainPrint { text }) => {
                Some(Event::Print(Print::message(text)))
            }

            ServerMessage::RoomUpdate(update) => {
                Some(self.update_room(update).unwrap_or_else(Event::Error))
            }

            ServerMessage::ReceivedItems(ReceivedItems { index, items }) => {
                if index == 0 {
                    self.received_items.clear();
                } else if index > self.received_items.len() {
                    // If the index of the item we just received doesn't
                    // match our expectation, follow the client
                    // guidelines and send a sync message requesting
                    // that the server send us all items over again.
                    return self.sync().err().map(Event::Error);
                }

                let receiver = &self.players[&self.player_key];
                let receiver_game = self.this_game();
                let items_or_err = items
                    .into_iter()
                    .enumerate()
                    .map(|(i, network)| {
                        let sender = self.teammate_arc(network.player)?;
                        let sender_game = self.game_or_err(sender.game())?;
                        let item = LocatedItem::hydrate_with_games(
                            network,
                            sender,
                            receiver.clone(),
                            sender_game,
                            receiver_game,
                        )?;
                        Ok(ReceivedItem::new(item, index + i))
                    })
                    .collect::<Result<Vec<ReceivedItem>, Error>>();

                Some(match items_or_err {
                    Ok(items) => {
                        self.received_items.extend(items);
                        Event::ReceivedItems(index)
                    }
                    Err(err) => Event::Error(err),
                })
            }

            ServerMessage::LocationInfo(LocationInfo { locations }) => {
                let sender = &self.players[&self.player_key];
                let sender_game = self.this_game();

                let locations_or_err = locations
                    .into_iter()
                    .map(|network| {
                        let receiver = self.teammate_arc(network.player)?;
                        let receiver_game = self.game_or_err(receiver.game())?;
                        LocatedItem::hydrate_with_games(
                            network,
                            sender.clone(),
                            receiver,
                            sender_game,
                            receiver_game,
                        )
                    })
                    .collect::<Result<Vec<LocatedItem>, Error>>();
                if let Some(sender) = self.location_scout_senders.pop_front() {
                    mem::drop(sender.send(locations_or_err));
                    None
                } else {
                    Some(Event::Error(
                        ProtocolError::ResponseWithoutRequest("LocationInfo").into(),
                    ))
                }
            }

            ServerMessage::Bounced(Bounced {
                games,
                slots,
                tags,
                data: BounceData::Generic(data),
            }) => Some(Event::Bounce {
                games,
                slots,
                tags,
                data,
            }),

            ServerMessage::Bounced(Bounced {
                games,
                slots,
                tags,
                data: BounceData::DeathLink(data),
            }) => Some(Event::DeathLink {
                games,
                slots,
                tags: tags.unwrap(),
                // We assume other clients try to normalize the time to
                // be the time on the server when the death occurred, so
                // add the server delay to translate that back into the
                // time on the client.
                time: data.time + self.server_skew,
                cause: data.cause,
                source: data.source,
            }),

            ServerMessage::InvalidPacket(InvalidPacket { text }) => {
                Some(Event::Error(Error::InvalidPacket(text)))
            }

            ServerMessage::Retrieved(Retrieved { keys }) => {
                if let Some(sender) = self.get_senders.pop_front() {
                    mem::drop(sender.send(Ok(keys)));
                    None
                } else {
                    Some(Event::Error(
                        ProtocolError::ResponseWithoutRequest("Retrieved").into(),
                    ))
                }
            }

            ServerMessage::SetReply(SetReply {
                key,
                value,
                original_value,
                slot: None,
            }) => Some(Event::KeyChanged {
                key,
                old_value: original_value,
                new_value: value,
                player: None,
            }),

            ServerMessage::SetReply(SetReply {
                key,
                value,
                original_value,
                slot: Some(slot),
            }) => Some(match self.teammate_arc(slot) {
                Ok(player) => Event::KeyChanged {
                    key,
                    old_value: original_value,
                    new_value: value,
                    player: Some(player),
                },
                Err(err) => Event::Error(err),
            }),

            ServerMessage::RoomInfo(_) => Some(Event::Error(
                ProtocolError::ResponseWithoutRequest("RoomInfo").into(),
            )),

            ServerMessage::ConnectionRefused(_) => Some(Event::Error(
                ProtocolError::ResponseWithoutRequest("ConnectionRefused").into(),
            )),

            ServerMessage::Connected(_) => Some(Event::Error(
                ProtocolError::ResponseWithoutRequest("Connected").into(),
            )),

            ServerMessage::DataPackage(_) => Some(Event::Error(
                ProtocolError::ResponseWithoutRequest("DataPackage").into(),
            )),
        }
    }

    /// Updates the room with the information in [update].
    fn update_room(&mut self, update: RoomUpdate) -> Result<Event, Error> {
        // Check for errors before making any changes so we don't end up in an
        // intermediate state.
        let checked_locations = update
            .checked_locations
            .map(|ids| {
                let game = self.this_game();
                ids.into_iter()
                    .map(|id| game.location_or_err(id))
                    .collect::<Result<Vec<_>, Error>>()
            })
            .transpose()?;

        let updated_players = update
            .players
            .map(|players| {
                players
                    .into_iter()
                    .filter_map(|new| {
                        self.players
                            .get(&(new.team, new.slot))
                            .ok_or(ProtocolError::MissingPlayer {
                                team: new.team,
                                slot: new.slot,
                            })
                            .map(|old| old.with_alias(new.alias))
                            .transpose()
                    })
                    .collect::<Result<Vec<_>, ProtocolError>>()
            })
            .transpose()?;

        let mut updated = Vec::new();
        if let Some(tags) = update.tags {
            updated.push(UpdatedField::ServerTags(mem::replace(
                &mut self.server_tags,
                tags,
            )))
        }

        if let Some(permissions) = update.permissions {
            updated.push(UpdatedField::Permissions {
                release: self.permissions.release,
                collect: self.permissions.collect,
                remaining: self.permissions.remaining,
            });
            self.permissions = permissions;
        }

        if update.hint_cost.is_some() || update.location_check_points.is_some() {
            updated.push(UpdatedField::HintEconomy {
                points_per_hint: self.points_per_hint(),
                hint_points_per_check: self.hint_points_per_check(),
            });
            if let Some(hint_cost_percentage) = update.hint_cost {
                self.hint_cost_percentage = hint_cost_percentage;
            }
            if let Some(hint_points_per_check) = update.location_check_points {
                self.hint_points_per_check = hint_points_per_check;
            }
        }

        if let Some(hint_points) = update.hint_points {
            updated.push(UpdatedField::HintPoints(mem::replace(
                &mut self.hint_points,
                hint_points,
            )))
        }

        if let Some(players) = updated_players {
            updated.push(UpdatedField::Players(
                players
                    .into_iter()
                    .filter_map(|p| self.players.insert((p.team(), p.slot()), p.into()))
                    .collect(),
            ))
        }

        if let Some(locations) = checked_locations {
            updated.push(UpdatedField::CheckedLocations(
                locations
                    .into_iter()
                    // Omit locations that we already know are checked from
                    // local information.
                    .filter(|loc| !self.local_locations_checked.insert(loc.id(), true).unwrap())
                    .collect(),
            ))
        }

        Ok(Event::Updated(updated))
    }
}

// The only reason Client doesn't automatically implement [Unpin] is that S
// might not implement it (although being decoded from JSON it probably does).
// Since we treat slot data as immutable anyway, we can guarantee that nothing
// will change and so it's safe to declare the entire Client as Unpin.
impl<S> Unpin for Client<S> where S: DeserializeOwned + 'static {}

// Safety: This isn't automatically Send due to `*const Game`, but that's just a
// pointer to data the client owns.
unsafe impl<S> Send for Client<S> where S: DeserializeOwned + Send + 'static {}
