use bytes::{Buf, BufMut, BytesMut};

use std::collections::{HashMap, HashSet};
use std::io::{BufRead, Cursor, Read};

use super::replay::*;
use super::shortcuts;
use crate::reader::{ReplayBufReadExt, ReplayReadError, ReplayReadExt, ReplayResult};

/// Configuration options for [`Parser`].
///
/// [`Parser`]: struct.Parser.html
#[derive(Clone, Debug)]
pub struct ParserOptions {
    /// Which commands should be fully parsed
    pub commands: HashSet<u8>,
    /// The maximum number of commands to parse
    pub limit: Option<usize>,
    /// Whether or not to add the parsed commands to the replay body
    pub save_commands: bool,
    /// Whether or not to return an error if a desync is detected
    pub stop_on_desync: bool,
}

/// A configurable replay parser. Construct a new parser with default paramaters or use the
/// [`ParserBuilder`] for fine grained control.
///
/// # Example
///
/// ```rust
/// use faf_replay_parser::scfa::*;
/// use std::io::Read;
///
/// let mut data: &[u8] = b"Some replay data";
///
/// let parser = Parser::new();
/// let result = parser.parse(&mut data)
///     .expect_err("Sadly not a valid replay");
/// ```
///
/// [`ParserBuilder`]: struct.ParserBuilder.html
#[derive(Debug)]
pub struct Parser {
    options: ParserOptions,
}

impl Parser {
    /// Constructs a new parser with default settings. This will only parse the most essential
    /// commands needed to determine the number of participants, the length of the replay, and
    /// whether or not the replay desynced.
    pub fn new() -> Parser {
        ParserBuilder::new().commands_default().build()
    }

    /// Constructs a new `StreamParser` with given [`ParserOptions`]
    pub fn with_options(options: ParserOptions) -> Parser {
        Parser { options }
    }

    /// Creates a new [`StreamParser`] with the same options as `self`.
    ///
    /// [`StreamParser`] struct.StreamParser.html
    pub fn new_stream(&self) -> StreamParser {
        StreamParser::with_options(self.options.clone())
    }

    /// Fully parses a stream of bytes into a `Replay` struct.
    pub fn parse(&self, reader: &mut (impl Read + BufRead)) -> ReplayResult<Replay> {
        self.parse_with_callback(reader, process_command)
    }

    /// Like [`parse`] but using a custom command processing function. This can be useful for
    /// aggregating replay results on the fly without saving the whole command stream to a `Vec`
    /// first.
    ///
    /// # Example
    /// ```rust
    /// use faf_replay_parser::scfa::*;
    ///
    /// let mut data: &[u8] = b"Some replay data";
    ///
    /// let parser = ParserBuilder::new()
    ///     .save_commands(false)
    ///     .build();
    ///
    /// let replay = parser.parse_with_callback(&mut data, |sim, command| {
    ///     if let ReplayCommand::Advance { ticks } = command {
    ///         println!("Advancing {}!", ticks);
    ///     }
    ///     Ok(())
    /// });
    /// ```
    ///
    /// [`parse`]: struct.Parser.html#method.parse
    pub fn parse_with_callback(
        &self,
        reader: &mut (impl Read + BufRead),
        callback: impl Fn(&mut SimData, &ReplayCommand) -> ReplayResult<()>,
    ) -> ReplayResult<Replay> {
        let mut buf = Vec::new();
        Ok(Replay {
            header: parse_header(reader, &mut buf)?,
            body: parse_body_with_callback(reader, &self.options, callback, &mut buf)?,
        })
    }

    /// Parses a stream of bytes into a `ReplayHeader` struct.
    pub fn parse_header(&self, reader: &mut (impl Read + BufRead)) -> ReplayResult<ReplayHeader> {
        Ok(parse_header(reader, &mut vec![])?)
    }

    /// Parses a stream of bytes into a `ReplayBody` struct. Usually this means the header has
    /// already been parsed so that `reader` starts at the correct offset for the replay body.
    pub fn parse_body(&self, reader: &mut impl Read) -> ReplayResult<ReplayBody> {
        self.parse_body_with_callback(reader, process_command)
    }

    /// Like [`parse_body`] but using a custom command processing function. See
    /// [`parse_with_callback`] for an example callback function.
    ///
    /// [`parse_body`]: struct.Parser.html#method.parse_body
    /// [`parse_with_callback`]: struct.Parser.html#method.parse_with_callback
    pub fn parse_body_with_callback(
        &self,
        reader: &mut impl Read,
        callback: impl Fn(&mut SimData, &ReplayCommand) -> ReplayResult<()>,
    ) -> ReplayResult<ReplayBody> {
        parse_body_with_callback(reader, &self.options, callback, &mut vec![])
    }
}

/// Like [`Parser`] but for incremental parsing of streamed data. Useful when data is being
/// received over a network.
///
/// Unlike [`Parser`], a `StreamParser` should only be used to parse a single replay stream at a
/// time.
///
/// # Example
/// ```rust
/// use faf_replay_parser::{StreamParser, ReplayReadError};
/// use std::io::Read;
/// use std::fs::File;
///
/// let mut file = File::open("tests/data/6176549.scfareplay").unwrap();
/// let mut stream = StreamParser::new();
///
/// while stream.feed_reader(&mut file, 8192).unwrap() != 0 {
///     match stream.parse() {
///         Err(ReplayReadError::IO(_)) => continue,
///         res => res.expect("Replay file ok")
///     }
/// }
/// let replay = stream.finalize().unwrap();
/// ```
#[derive(Debug)]
pub struct StreamParser {
    options: ParserOptions,
    /// Captures incoming unparsed data.
    buffer: BytesMut,
    /// A re-usable buffer for reading command data.
    reuse_buf: Vec<u8>,
    header: Option<ReplayHeader>,
    body: Option<ReplayBody>,
}

impl StreamParser {
    /// Constructs a new `StreamParser` with default settings.
    /// See [`ParserBuilder::commands_default`].
    pub fn new() -> StreamParser {
        ParserBuilder::new().commands_default().build_stream()
    }

    /// Constructs a new `StreamParser` with given [`ParserOptions`].
    pub fn with_options(options: ParserOptions) -> StreamParser {
        StreamParser {
            options,
            buffer: BytesMut::new(),
            reuse_buf: Vec::new(),
            header: None,
            body: None,
        }
    }

    /// Returns a reference to the stored [`ReplayHeader`].
    pub fn header(&self) -> Option<&ReplayHeader> {
        self.header.as_ref()
    }

    /// Returns a reference to the stored [`ReplayBody`].
    pub fn body(&self) -> Option<&ReplayBody> {
        self.body.as_ref()
    }

    /// Adds some replay data to the buffer from an existing slice.
    ///
    /// If data is coming from a [`std::io::Read`], then
    /// [`feed_reader`](struct.StreamParser.html#method.feed_reader) should be used instead to
    /// avoid double allocation and double copying.
    ///
    /// # Examples
    /// ```rust
    /// use faf_replay_parser::StreamParser;
    ///
    /// let mut stream = StreamParser::new();
    /// stream.feed(&[0u8; 10]);
    /// ```
    pub fn feed(&mut self, data: &[u8]) {
        self.buffer.extend_from_slice(data)
    }

    /// Pulls data from a [`std::io::Read`] and adds it directly to the buffer. This will only make
    /// one call to `read` reading up to `max` bytes. Calling this in a loop acts similar to a
    /// [`std::io::BufReader`] with a buffer size of `max`. Setting `max` to a small number could
    /// cause excessive calls to `read`.
    ///
    /// # Returns
    /// The number of bytes read.
    ///
    /// # Examples
    /// ```rust
    /// use faf_replay_parser::StreamParser;
    /// use std::io::Cursor;
    ///
    /// let mut stream = StreamParser::new();
    /// let mut reader = Cursor::new(vec![0u8; 10]);
    ///
    /// assert_eq!(stream.feed_reader(&mut reader, 3).unwrap(), 3);
    /// assert_eq!(stream.feed_reader(&mut reader, 10).unwrap(), 7);
    /// ```
    pub fn feed_reader(&mut self, reader: &mut impl Read, max: usize) -> std::io::Result<usize> {
        // TODO: Should this even have a `max` option? Using a small number could be inefficient.
        self.buffer.reserve(max);
        let uninit_slice = self.buffer.chunk_mut();

        unsafe {
            let buf = std::slice::from_raw_parts_mut(uninit_slice.as_mut_ptr(), max);
            // `read` may read the contents of buf, so we need to initialize it.
            // There is a nightly feature `read_initializer` that can be used to avoid this.
            buf.fill(0);

            let n = reader.read(buf)?;

            self.buffer.advance_mut(n);
            Ok(n)
        }
    }

    /// Parse as much of the data in the buffer as possible and advance the internal state. Calling
    /// this function a second time without calling [`feed`](struct.StreamParser.html#method.feed)
    /// or [`feed_reader`](struct.StreamParser.html#method.feed_reader) will return an error.
    ///
    /// # Errors
    /// Returns a [`ReplayReadError::IO`] if there is not enough data available, and other variants
    /// of [`ReplayReadError`] if the data stream is corrupt.
    pub fn parse(&mut self) -> ReplayResult<()> {
        if self.header.is_none() {
            self.parse_header()
        } else {
            self.parse_body()
        }
    }

    /// Parse a [`ReplayHeader`] from the buffer and advance the internal state. If `Ok(())` is
    /// returned, the header can be accessed by calling
    /// [`header`](struct.StreamParser.html#method.header).
    ///
    /// # Errors
    /// Returns a [`ReplayReadError::IO`] if there is not enough data available, and other variants
    /// of [`ReplayReadError`] if the data stream is corrupt.
    pub fn parse_header(&mut self) -> ReplayResult<()> {
        let mut cur = Cursor::new(&self.buffer[..]);
        let result = parse_header(&mut cur, &mut self.reuse_buf);

        let result = match result {
            Err(ReplayReadError::IO(_)) => {
                // IO operations on BytesMut are infallable, so we must be missing data.
                // In this case we just return the error without advancing the buffer, and let the
                // caller try again later.
                return Err(result.unwrap_err());
            }
            Ok(header) => {
                self.header.replace(header);
                Ok(())
            }
            err => Err(err.unwrap_err()),
        };

        let n = cur.position() as usize;
        self.buffer.advance(n);

        return result;
    }

    /// Parse as many commands as possible from the buffer and add them to the internal
    /// [`ReplayBody`] command stream. The [`ReplayBody`] can be accessed by calling
    /// [`body`](struct.StreamParser.html#method.body).
    ///
    /// # Errors
    /// Returns a [`ReplayReadError::IO`] if there is not enough data available, and other variants
    /// of [`ReplayReadError`] if the data stream is corrupt.
    pub fn parse_body(&mut self) -> ReplayResult<()> {
        self.parse_body_with_callback(process_command)
    }

    /// Like [`parse_body`](struct.StreamParser.html#method.parse_body) but applying a custom
    /// processing function to each parsed command.
    pub fn parse_body_with_callback(
        &mut self,
        callback: impl Fn(&mut SimData, &ReplayCommand) -> ReplayResult<()>,
    ) -> ReplayResult<()> {
        self.body.get_or_insert_with(|| ReplayBody {
            commands: Vec::new(),
            sim: SimData::new(),
        });

        while self.options.limit.is_none()
            || self.body.as_ref().unwrap().commands.len() < self.options.limit.unwrap()
        {
            // This duplicates a small amount of work, but saves one allocation.
            if !self.has_frame()? {
                return Err(ReplayReadError::IO(std::io::Error::new(
                    std::io::ErrorKind::UnexpectedEof,
                    "partial frame",
                )));
            }
            match self.parse_command() {
                Ok(Some(cmd)) => {
                    let result = callback(&mut self.body.as_mut().unwrap().sim, &cmd);

                    // Deciding not to combine these with #![feature(let_chains)]
                    if let Err(ReplayReadError::Desynced(_)) = result {
                        if self.options.stop_on_desync {
                            result?;
                        }
                    } else {
                        result?;
                    }

                    if self.options.save_commands {
                        self.body.as_mut().unwrap().commands.push(cmd);
                    }

                    if self.buffer.is_empty() {
                        return Ok(());
                    }
                }
                Ok(None) => {}
                err => return Err(err.unwrap_err()),
            }
        }

        // Skip commands without parsing
        loop {
            self.parse_command_frame().map(|_| ())?;
            if self.buffer.is_empty() {
                return Ok(());
            }
        }
    }

    /// Checks if there is enough data available to parse an additional command from the replay
    /// body.
    ///
    /// # Errors
    /// If the command ID or the frame size are invalid, a [`ReplayReadError::Malformed`] is
    /// returned.
    ///
    /// # Examples
    /// ```rust
    /// use faf_replay_parser::StreamParser;
    ///
    /// let mut stream = StreamParser::new();
    /// assert!(!stream.has_frame().unwrap());
    ///
    /// stream.feed(&[0, 3, 0]);
    /// assert!(stream.has_frame().unwrap());
    ///
    /// stream.reset();
    /// stream.feed(&[100]); // Not a valid command id
    /// stream.has_frame().unwrap_err();
    /// ```
    pub fn has_frame(&self) -> ReplayResult<bool> {
        shortcuts::has_frame(&self.buffer[..])
    }

    /// Parses a command from the buffer and consumes the read data. If the command type is not
    /// present in the parser options then returns `Ok(None)`.
    ///
    /// It is recommended to call [`has_frame`](struct.StreamParser.html#method.has_frame) first to
    /// ensure that enough data is available.
    ///
    /// # Errors
    /// If an error occurrs, it may be that the replay data is corrupted or there is not enough
    /// data available to fully parse a command. In case of missing data, the error will always
    /// be a [`ReplayReadError::IO`] and the inner error will be a [`std::io::Error`] with kind
    /// [`std::io::ErrorKind::UnexpectedEof`]. Any other error likely means that the stream became
    /// corrupted and that future calls to `parse_command` are unlikely to succeed.
    ///
    /// # Examples
    /// ```rust
    /// use faf_replay_parser::StreamParser;
    /// use faf_replay_parser::scfa::ReplayCommand;
    ///
    /// let mut stream = StreamParser::new();
    ///
    /// stream.feed(&[0x00, 0x07, 0x00, 0x01, 0x00, 0x00, 0x00]);
    ///
    /// assert_eq!(
    ///     stream.parse_command().unwrap(),
    ///     Some(ReplayCommand::Advance { ticks: 1 })
    /// );
    /// ```
    pub fn parse_command(&mut self) -> ReplayResult<Option<ReplayCommand>> {
        // Might be able to do this using BytesMut functionality alone. Not sure.
        let mut cur = Cursor::new(&self.buffer[..]);
        let result = parse_command(&mut cur, &self.options, &mut self.reuse_buf);

        if let Err(ReplayReadError::IO(_)) = &result {
            // IO operations on BytesMut are infallable, so we must be missing data.
            // In this case we just return the error without advancing the buffer, and let the
            // caller try again later.
            return result;
        } else {
            let n = cur.position() as usize;
            self.buffer.advance(n);

            return result;
        }
    }

    /// Parses a command frame, but doesn't parse the command data.
    ///
    /// # Errors
    /// Returns a [`ReplayReadError::IO`] if there is not enough data available, and other variants
    /// of [`ReplayReadError`] if either the command id or the frame size are invalid.
    pub fn parse_command_frame(&mut self) -> ReplayResult<ReplayCommandFrame> {
        let mut cur = Cursor::new(&self.buffer[..]);
        let result = parse_command_frame(&mut cur);

        if let Err(ReplayReadError::IO(_)) = &result {
            return result;
        } else {
            let n = cur.position() as usize;
            self.buffer.advance(n);

            return result;
        }
    }

    /// Returns whether or not a call to [`finalize`](struct.StreamParser.html#method.finalize)
    /// will succeed.
    pub fn can_finalize(&self) -> bool {
        self.buffer.is_empty()
    }

    /// Signals that the stream has ended and returns the parsed [`StreamReplay`].
    ///
    /// # Errors
    /// Returns a [`ReplayReadError::IO`] if there is any unparsed data in the buffer.
    pub fn finalize(&mut self) -> ReplayResult<StreamReplay> {
        match self.buffer.is_empty() {
            true => Ok(self.force_finalize()),
            false => Err(ReplayReadError::IO(std::io::Error::new(
                std::io::ErrorKind::UnexpectedEof,
                "replay finalized before all data was consumed",
            ))),
        }
    }

    /// Returns the currently parsed replay, and discards any excess data from the buffer.
    pub fn force_finalize(&mut self) -> StreamReplay {
        let result = StreamReplay {
            header: self.header.take(),
            body: self.body.take(),
        };
        self.buffer.clear();

        result
    }

    /// Throws away all internal state. After resetting, `self` can be used to start parsing a new
    /// data stream.
    ///
    /// Note that this will not deallocate the internal buffer, so it is slightly more efficient to
    /// reset and reuse a [`StreamParser`] than to create a new one.
    pub fn reset(&mut self) {
        self.header.take();
        self.body.take();
        self.buffer.clear();
    }
}

/// Like [`Replay`] but fields are optional.
#[derive(Debug)]
pub struct StreamReplay {
    pub header: Option<ReplayHeader>,
    pub body: Option<ReplayBody>,
}

impl StreamReplay {
    /// Unwrap the optional fields.
    pub fn unwrap(self) -> Replay {
        Replay {
            header: self.header.unwrap(),
            body: self.body.unwrap(),
        }
    }
}

/// A builder for configuring replay [`Parser`]s. See [`build`] for an example.
///
/// [`Parser`]: struct.Parser.html
/// [`build`]: struct.ParserBuilder.html#method.build
#[derive(Debug)]
pub struct ParserBuilder {
    options: ParserOptions,
}

impl ParserBuilder {
    /// Initialize with default options
    pub fn new() -> ParserBuilder {
        ParserBuilder {
            options: ParserOptions {
                commands: HashSet::new(),
                limit: None,
                save_commands: true,
                stop_on_desync: true,
            },
        }
    }

    /// Creates a new [`Parser`] by moving the internal `ParserOptions` and consuming self.
    ///
    /// # Example
    ///
    /// ```rust
    /// use faf_replay_parser::scfa::*;
    ///
    /// let parser = ParserBuilder::new()
    ///     .commands(&[
    ///         replay_command::ADVANCE,
    ///         replay_command::VERIFY_CHECKSUM,
    ///         replay_command::END_GAME
    ///     ])
    ///     .build();
    /// ```
    ///
    /// [`Parser`]: struct.Parser.html
    pub fn build(self) -> Parser {
        Parser::with_options(self.options)
    }

    /// Like [`build`] but creates a new [`StreamParser`].
    ///
    /// [`StreamParser`]: struct.StreamParser.html
    /// [`build`]: struct.ParserBuilder.html#method.build
    pub fn build_stream(self) -> StreamParser {
        StreamParser::with_options(self.options)
    }

    /// Creates a new `Parser` by cloning the internal `ParserOptions`. Useful when constructing
    /// many parsers with similar options.
    ///
    /// Example
    ///
    /// ```rust
    /// use faf_replay_parser::scfa::*;
    ///
    /// let mut builder = ParserBuilder::new()
    ///     .commands(&[
    ///         replay_command::ADVANCE,
    ///         replay_command::END_GAME
    ///     ]);
    ///
    /// // Can only count game ticks
    /// let tick_parser = builder.build_clone();
    ///
    /// let builder = builder.command(replay_command::VERIFY_CHECKSUM);
    /// // Can count game ticks and detect desyncs
    /// let desync_parser = builder.build_clone();
    ///
    /// let builder = builder.command(replay_command::LUA_SIM_CALLBACK);
    /// // Can count game ticks, detect desyncs, and extract chat messages
    /// let message_parser = builder.build();
    /// ```
    pub fn build_clone(&self) -> Parser {
        Parser::with_options(self.options.clone())
    }

    /// Like [`build_clone`] but creates a new [`StreamParser`].
    ///
    /// [`StreamParser`]: struct.StreamParser.html
    /// [`build_clone`]: struct.ParserBuilder.html#method.build_clone
    pub fn build_stream_clone(&self) -> StreamParser {
        StreamParser::with_options(self.options.clone())
    }

    /// Add a command to the `ParserOptions`.
    pub fn command(mut self, command: u8) -> ParserBuilder {
        self.options.commands.insert(command);
        self
    }

    /// Add multiple commands to the `ParserOptions`.
    pub fn commands(mut self, commands: &[u8]) -> ParserBuilder {
        commands.iter().for_each(|cmd| {
            self.options.commands.insert(*cmd);
        });
        self
    }

    /// Add all supported commands to the `ParserOptions`.
    pub fn commands_all(self) -> ParserBuilder {
        self.commands((0..=replay_command::MAX).collect::<Vec<u8>>().as_slice())
    }

    /// Add the default set of commands to the `ParserOptions`.
    ///
    /// Default commands are `ADVANCE`, `SET_COMMAND_SOURCE`, `COMMAND_SOURCE_TERMINATED`,
    /// `VERIFY_CHECKSUM`, and `END_GAME`
    pub fn commands_default(self) -> ParserBuilder {
        self.commands(&[
            replay_command::ADVANCE,
            replay_command::SET_COMMAND_SOURCE,
            replay_command::COMMAND_SOURCE_TERMINATED,
            replay_command::VERIFY_CHECKSUM,
            replay_command::END_GAME,
        ])
    }

    /// Set a limit on the number of commands to parse. This only applies to commands in the
    /// `ParserOptions.commands` field.
    pub fn limit(mut self, limit: Option<usize>) -> ParserBuilder {
        self.options.limit = limit;
        self
    }

    /// Whether or not to store the parsed commands in the parsed `ReplayBody`. When set to false,
    /// `Replay.body.commands.len()` will always be `0`.
    pub fn save_commands(mut self, save_commands: bool) -> ParserBuilder {
        self.options.save_commands = save_commands;
        self
    }

    /// Whether or not to return an error if a desync is detected. When set to false,
    /// `Replay.sim.desync_ticks` will be a list of all desynced ticks.
    pub fn stop_on_desync(mut self, stop_on_desync: bool) -> ParserBuilder {
        self.options.stop_on_desync = stop_on_desync;
        self
    }
}

/// Like `shortcuts::body_ticks` but for generic readers.
pub fn parse_body_ticks(reader: &mut impl Read) -> ReplayResult<u32> {
    let mut ticks = 0;
    let mut buf = vec![0u8; 4096];

    // Will not throw an error on missing data
    loop {
        unsafe {
            ticks += match parse_body_ticks_command(reader, buf.as_mut_slice()) {
                Err(ReplayReadError::IO(err))
                    if err.kind() == std::io::ErrorKind::UnexpectedEof =>
                {
                    return Ok(ticks);
                }
                other => other?,
            }
        }
    }
}

#[inline]
/// Calling this with a buffer shorter than 4 bytes is undefined behavior.
unsafe fn parse_body_ticks_command(reader: &mut impl Read, buf: &mut [u8]) -> ReplayResult<u32> {
    debug_assert!(buf.len() >= 4);

    // Size is guaranteed to be at least 3.
    let (command_id, size) = parse_command_frame_header(reader)?;

    if command_id != replay_command::ADVANCE {
        reader.skip_buf(size as usize - 3, buf)?;
        return Ok(0);
    }
    if size != 7 {
        reader.skip_buf(size as usize - 3, buf)?;
        return Err(ReplayReadError::Malformed("invalid command size"));
    }
    // We are assuming that buf is large enough to hold the contents of an ADVANCE command frame.
    reader.read_exact(buf.get_unchecked_mut(0..4))?;
    let ticks = buf.get_unchecked(0..4).read_u32_le().unwrap();

    Ok(ticks)
}

fn parse_header(data: &mut (impl Read + BufRead), buf: &mut Vec<u8>) -> ReplayResult<ReplayHeader> {
    // Format will be very close to "Supreme Commander v1.50.3701"
    let scfa_version = data.read_string_with_capacity(28)?;
    // Skip the string "\r\n\x00"
    // Even though we expect exactly this string, we still allow for longer strings just in case.
    data.read_string()?;
    let version_and_mapname = data.read_string()?;
    let mut version_and_mapname = version_and_mapname.splitn(2, "\r\n");
    let replay_version = match version_and_mapname.next() {
        Some(s) => s.to_string(),
        None => return Err(ReplayReadError::Malformed("missing replay version")),
    };
    let map_file = match version_and_mapname.next() {
        Some(s) => s.to_string(),
        None => return Err(ReplayReadError::Malformed("missing map name")),
    };
    // Skip the string "\r\n\x1a\x00"
    data.read_string()?;

    let mods_size = data.read_u32_le()? as usize;
    data.read_exact_to_vec(mods_size, buf)?;
    let mods = (&buf[..mods_size]).read_lua_object()?;

    let scenario_size = data.read_u32_le()? as usize;
    data.read_exact_to_vec(scenario_size, buf)?;
    let scenario = (&buf[..scenario_size]).read_lua_object()?;
    let num_sources = data.read_u8()? as usize;

    let mut players = HashMap::new();
    for _ in 0..num_sources {
        let name = data.read_string()?;
        let player_id = data.read_u32_le()?;
        players.insert(name, player_id);
    }

    let cheats_enabled = data.read_bool()?;
    let army_count = data.read_u8()? as usize;

    let mut armies = HashMap::new();
    for _ in 0..army_count {
        let player_data_size = data.read_u32_le()? as usize;
        data.read_exact_to_vec(player_data_size, buf)?;
        let player_data = (&buf[..player_data_size]).read_lua_object()?;
        let player_source = data.read_u8()? as u32;
        armies.insert(player_source, player_data);

        if player_source != 255 {
            data.skip(1)?;
        }
    }

    let seed = data.read_u32_le()?;

    Ok(ReplayHeader {
        scfa_version,
        replay_version,
        map_file,
        mods,
        scenario,
        players,
        cheats_enabled,
        army_count,
        armies,
        seed,
    })
}

/// Parse replay command stream
fn parse_body_with_callback(
    reader: &mut impl Read,
    options: &ParserOptions,
    callback: impl Fn(&mut SimData, &ReplayCommand) -> ReplayResult<()>,
    buf: &mut Vec<u8>,
) -> ReplayResult<ReplayBody> {
    let mut commands = Vec::new();
    let mut sim = SimData::new();

    while options.limit.is_none() || commands.len() < options.limit.unwrap() {
        match parse_command(reader, options, buf) {
            Err(ReplayReadError::IO(ref e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                break
            }
            Err(e) => return Err(e),
            Ok(opt) => {
                if let Some(cmd) = opt {
                    let result = callback(&mut sim, &cmd);

                    // Deciding not to combine these with #![feature(let_chains)]
                    if let Err(ReplayReadError::Desynced(_)) = result {
                        if options.stop_on_desync {
                            result?;
                        }
                    } else {
                        result?;
                    }

                    if options.save_commands {
                        commands.push(cmd);
                    }
                }
            }
        }
    }

    Ok(ReplayBody { commands, sim })
}

fn parse_command_frame(reader: &mut (impl Read + BufRead)) -> ReplayResult<ReplayCommandFrame> {
    let (command_id, size) = parse_command_frame_header(reader)?;
    let data = reader.vec_read_exact((size - 3) as usize)?;

    Ok(ReplayCommandFrame {
        command_id,
        size,
        data,
    })
}

fn parse_command_frame_header(reader: &mut impl Read) -> ReplayResult<(u8, u16)> {
    let command_id = reader.read_u8()?;
    if command_id > replay_command::MAX {
        return Err(ReplayReadError::Malformed("invalid command"));
    }

    let size = reader.read_u16_le()?;
    if size < 3 {
        return Err(ReplayReadError::Malformed("invalid command size"));
    }

    Ok((command_id, size))
}

/// Parses one command
/// Packet structure in bytestream
/// ::
///     4   7      7      7
///     TLLDTLLDDDDTLLDDDDTLLDDDD
///     =========================
/// Where:
/// ::
///     T - byte - defines command type
///     L - short - defines command length of T + L + D
///     D - variable length - binary data of size `command length`, may be empty
pub fn parse_command(
    reader: &mut impl Read,
    ctx: &ParserOptions,
    buf: &mut Vec<u8>,
) -> ReplayResult<Option<ReplayCommand>> {
    use replay_command::*;
    use ReplayCommand::*;

    let (command_id, size) = parse_command_frame_header(reader)?;
    let len = size as usize - 3;

    // Pull command data from the reader.
    reader.read_exact_to_vec(len, buf)?;

    if !ctx.commands.contains(&command_id) {
        return Ok(None);
    }
    let mut reader = &buf[..len];

    let ret = match command_id {
        ADVANCE => Advance {
            ticks: reader.read_u32_le()?,
        },
        SET_COMMAND_SOURCE => SetCommandSource {
            id: reader.read_u8()?,
        },
        COMMAND_SOURCE_TERMINATED => CommandSourceTerminated,
        VERIFY_CHECKSUM => VerifyChecksum {
            digest: reader.vec_read_exact(16)?,
            tick: reader.read_u32_le()?,
        },
        REQUEST_PAUSE => RequestPause,
        RESUME => Resume,
        SINGLE_STEP => SingleStep,
        CREATE_UNIT => CreateUnit {
            army: reader.read_u8()?,
            blueprint: reader.read_string()?,
            x: reader.read_f32_le()?,
            z: reader.read_f32_le()?,
            heading: reader.read_f32_le()?,
        },
        CREATE_PROP => CreateProp {
            blueprint: reader.read_string()?,
            position: read_position(&mut reader)?,
        },
        DESTROY_ENTITY => DestroyEntity {
            unit: reader.read_u32_le()?,
        },
        WARP_ENTITY => WarpEntity {
            unit: reader.read_u32_le()?,
            x: reader.read_f32_le()?,
            y: reader.read_f32_le()?,
            z: reader.read_f32_le()?,
        },
        PROCESS_INFO_PAIR => ProcessInfoPair {
            unit: reader.read_u32_le()?,
            arg1: reader.read_string()?,
            arg2: reader.read_string()?,
        },
        ISSUE_COMMAND => IssueCommand(read_game_command(&mut reader)?),
        ISSUE_FACTORY_COMMAND => IssueFactoryCommand(read_game_command(&mut reader)?),
        INCREASE_COMMAND_COUNT => IncreaseCommandCount {
            id: reader.read_u32_le()?,
            delta: reader.read_i32_le()?,
        },
        DECREASE_COMMAND_COUNT => DecreaseCommandCount {
            id: reader.read_u32_le()?,
            delta: reader.read_i32_le()?,
        },
        SET_COMMAND_TARGET => SetCommandTarget {
            id: reader.read_u32_le()?,
            target: read_target(&mut reader)?,
        },
        SET_COMMAND_TYPE => {
            let id = reader.read_u32_le()?;
            let type_ = reader.read_u32_le()? as u8;
            if type_ > game_command::MAX {
                return Err(ReplayReadError::Malformed("invalid game command"));
            }
            SetCommandType { id, type_ }
        }
        SET_COMMAND_CELLS => {
            // This function may or may not be correct. So far, we have not found any replays that
            // contain this command.
            let id = reader.read_u32_le()?;
            let cells = reader.read_lua_object()?;
            if cells.evaluate_as_bool() {
                reader.skip(1)?;
            }
            SetCommandCells {
                id,
                cells,
                position: read_position(&mut reader)?,
            }
        }
        REMOVE_COMMAND_FROM_QUEUE => RemoveCommandFromQueue {
            id: reader.read_u32_le()?,
            unit: reader.read_u32_le()?,
        },
        DEBUG_COMMAND => DebugCommand {
            command: reader.read_string()?,
            position: read_position(&mut reader)?,
            focus_army: reader.read_u8()?,
            selection: read_entity_list(&mut reader)?,
        },
        EXECUTE_LUA_IN_SIM => ExecuteLuaInSim {
            code: reader.read_string()?,
        },
        LUA_SIM_CALLBACK => {
            let func = reader.read_string()?;
            let args = reader.read_lua_object()?;

            let selection = read_entity_list(&mut reader)?;
            LuaSimCallback {
                func,
                args,
                selection,
            }
        }
        END_GAME => EndGame,
        _ => {
            return Err(ReplayReadError::Malformed(
                "encountered unrecognized command",
            ))
        }
    };

    // Make sure we've read all available data
    debug_assert!(reader.vec_read_to_end().unwrap().is_empty());

    Ok(Some(ret))
}

fn read_position(reader: &mut impl Read) -> ReplayResult<Position> {
    Ok(Position {
        x: reader.read_f32_le()?,
        y: reader.read_f32_le()?,
        z: reader.read_f32_le()?,
    })
}

fn read_entity_list(reader: &mut impl Read) -> ReplayResult<Vec<u32>> {
    let len = reader.read_u32_le()? as usize;
    // 16 players at max unit cap selecting every entity at once.
    // This should be way more than enough.
    let mut result = Vec::with_capacity(std::cmp::min(len, 16_000));

    for _ in 0..len {
        result.push(reader.read_u32_le()?);
    }

    Ok(result)
}

fn read_target(reader: &mut impl Read) -> ReplayResult<Target> {
    use target_type::*;

    match reader.read_u8()? {
        NONE => Ok(Target::None),
        ENTITY => Ok(Target::Entity {
            id: reader.read_u32_le()?,
        }),
        POSITION => Ok(Target::Position(read_position(reader)?)),
        _ => Err(ReplayReadError::Malformed("invalid target type")),
    }
}

pub fn read_formation(reader: &mut impl Read) -> ReplayResult<Option<Formation>> {
    let type_ = reader.read_i32_le()?;
    if type_ == -1 {
        return Ok(None);
    }

    Ok(Some(Formation {
        a: reader.read_f32_le()?,
        b: reader.read_f32_le()?,
        c: reader.read_f32_le()?,
        d: reader.read_f32_le()?,
        scale: reader.read_f32_le()?,
    }))
}

pub fn read_cells(reader: &mut impl Read) -> ReplayResult<Vec<Cell>> {
    let size = reader.read_u32_le()? as usize;
    // Since we've never seen these, lets just limit to 1 MiB
    let mut result = Vec::with_capacity(std::cmp::min(
        size,
        crate::MIB / std::mem::size_of::<Cell>(),
    ));
    for _ in 0..size {
        result.push(Cell {
            x: reader.read_i16_le()?,
            z: reader.read_i16_le()?,
        });
    }
    Ok(result)
}

pub fn read_game_command(reader: &mut (impl Read + BufRead)) -> ReplayResult<GameCommand> {
    let entity_ids = read_entity_list(reader)?;
    let id = reader.read_u32_le()?;

    // Guessing that 0xFFFFFFFF for this field means it's a signed int
    // Appears to be another command id referencing the target of a coordinated attack.
    // When this is not -1, type_ is always 'ATTACK (10)'
    let coordinated_attack_cmd_id = reader.read_u32_le()?;
    let type_ = reader.read_u8()?;

    if type_ > game_command::MAX {
        return Err(ReplayReadError::Malformed("invalid game command"));
    }
    // Appears to be a command id or entity id
    // When this is not -1, type_ is always "BUILD_FACTORY (7)"
    let arg2 = reader.read_i32_le()?;

    let target = read_target(reader)?;
    let arg3 = reader.read_u8()?; // Always 0
    let formation = read_formation(reader)?;

    let blueprint = reader.by_ref().take(255).read_string()?;

    // TODO: Are these signed or unsigned? What do they mean?
    let arg4 = reader.read_u32_le()?; // Always 0
    let arg5 = reader.read_u32_le()?; // Always 1
    let arg6 = reader.read_u32_le()?; // Always 1

    let mut clear_queue = None;

    let upgrades = reader.read_lua_object()?;
    if upgrades.evaluate_as_bool() {
        clear_queue = Some(reader.read_bool()?);
    }

    Ok(GameCommand {
        entity_ids,
        id,
        coordinated_attack_cmd_id,
        type_,
        arg2,
        target,
        arg3,
        formation,
        blueprint,
        arg4,
        arg5,
        arg6,
        upgrades,
        clear_queue,
    })
}

/// Update `sim` given the new command.
fn process_command(sim: &mut SimData, command: &ReplayCommand) -> ReplayResult<()> {
    use ReplayCommand::*;

    match command {
        // Increment the tick counter
        Advance { ticks } => sim.tick += *ticks,
        // Change which player the following commands apply to
        SetCommandSource { id } => sim.command_source = *id,
        // Record when command a command stream ends
        CommandSourceTerminated => {
            sim.players_last_tick.insert(sim.command_source, sim.tick);
        }
        // Make sure that the replay hasn't desynced. Every player sends a checksum once per 50
        // ticks representing the game state of their local simulation. If one or more of the
        // checksums don't match, then those players have desynced and we can no longer trust their
        // replay streams.
        VerifyChecksum { digest, tick } => {
            if sim.checksum_tick < Some(*tick) {
                // First checksum submitted for this tick
                sim.checksum_tick = Some(*tick);
                sim.checksum.copy_from_slice(&digest[..16]);
                return Ok(());
            }
            debug_assert_eq!(
                sim.checksum_tick,
                Some(*tick),
                "Checksum received out of order!"
            );

            if sim.checksum != digest[..16] {
                if sim.desync_tick.is_none() {
                    sim.desync_tick = Some(sim.tick);
                    sim.desync_ticks = Some(vec![]);
                }

                sim.desync_ticks.as_mut().unwrap().push(sim.tick);

                // TODO: Use a different error type here? ParseError?
                return Err(ReplayReadError::Desynced(sim.tick));
            }
        }
        _ => (),
    };
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::lua::LuaObject;
    use std::io::Write;

    #[test]
    fn parse_issue_command() {
        let mut data: &[u8] = &[
            // cmd header -- |  -- entities len -- |   -- entity id: 0 --  |   -- command id  ...
            0x0C, 0x40, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
            //  |      -- arg1 --       | type |      -- arg2 --      | ttype|     -- x (f32) ...
            0x00, 0xFF, 0xFF, 0xFF, 0xFF, 0x08, 0xFF, 0xFF, 0xFF, 0xFF, 0x02, 0x00, 0x60, 0x28,
            //-- |    -- y (f32)  --    |     -- z (f32) --     | arg3 | -- formation (i32) -- |
            0x44, 0x00, 0x70, 0x95, 0x41, 0x00, 0x40, 0xA9, 0x43, 0x00, 0xFF, 0xFF, 0xFF, 0xFF,
            //             -- blueprint id --             |        -- arg4 --     | -- arg5   ...
            0x75, 0x72, 0x62, 0x30, 0x31, 0x30, 0x31, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00,
            // arg5 -- |       -- arg6 --      | Nil |
            0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x02, 0x01, 0x01, 0x04, 0x00, 0x01,
        ];
        let ctx = ParserOptions {
            commands: [replay_command::ISSUE_COMMAND].iter().cloned().collect(),
            limit: None,
            save_commands: true,
            stop_on_desync: true,
        };

        let issue_command = parse_command(&mut data, &ctx, &mut vec![])
            .unwrap()
            .unwrap();

        // I'm so glad that Rust doesn't implement Eq for floats
        let command = GameCommand {
            entity_ids: vec![0],
            id: 0,
            coordinated_attack_cmd_id: std::u32::MAX,
            type_: game_command::BUILD_MOBILE,
            arg2: -1,
            target: Target::Position(Position {
                x: 0.0,
                y: 0.0,
                z: 0.0,
            }),
            arg3: 0,
            formation: None,
            blueprint: "urb0101".to_string(),
            arg4: 0,
            arg5: 1,
            arg6: 1,
            upgrades: LuaObject::Nil,
            clear_queue: None,
        };
        match issue_command {
            ReplayCommand::IssueCommand(issue_command) => {
                assert_eq!(command.entity_ids, issue_command.entity_ids);
                assert_eq!(command.id, issue_command.id);
                assert_eq!(
                    command.coordinated_attack_cmd_id,
                    issue_command.coordinated_attack_cmd_id
                );
                assert_eq!(command.type_, issue_command.type_);
                assert_eq!(command.arg2, issue_command.arg2);
                assert_eq!(command.arg3, issue_command.arg3);
                assert_eq!(command.blueprint, issue_command.blueprint);
                assert_eq!(command.arg4, issue_command.arg4);
                assert_eq!(command.arg5, issue_command.arg5);
                assert_eq!(command.arg6, issue_command.arg6);
                assert_eq!(command.upgrades, issue_command.upgrades);
                assert_eq!(command.clear_queue, issue_command.clear_queue);
            }
            _ => panic!("Wrong command type produced"),
        }
    }

    #[test]
    fn parse_lua_sim_callback() {
        let mut data: &[u8] = &[
            0x16, 0x1A, 0x00, 0x43, 0x6C, 0x65, 0x61, 0x72, 0x54, 0x61, 0x72, 0x67, 0x65, 0x74,
            0x73, 0x00, 0x04, 0x05, 0x01, 0x00, 0x00, 0x00, 0xE0, 0x00, 0x40, 0x00,
        ];
        let ctx = ParserOptions {
            commands: [replay_command::LUA_SIM_CALLBACK].iter().cloned().collect(),
            limit: None,
            save_commands: true,
            stop_on_desync: true,
        };

        let command = parse_command(&mut data, &ctx, &mut vec![])
            .unwrap()
            .unwrap();

        match command {
            ReplayCommand::LuaSimCallback {
                func,
                args,
                selection,
            } => {
                assert_eq!(func, "ClearTargets");
                assert!(args.as_hashmap().unwrap().is_empty());
                assert_eq!(selection, vec![0x4000E0]);
            }
            _ => panic!("Wrong command type produced"),
        }
    }

    #[test]
    fn test_stream_init() {
        let stream = StreamParser::new();

        assert_eq!(stream.header().is_none(), true);
        assert_eq!(stream.body().is_none(), true);
    }

    #[test]
    fn test_stream_parse_command() {
        let mut stream = StreamParser::new();

        assert_eq!(stream.has_frame().unwrap(), false);

        // Advance is a default command
        stream.feed(&[0x00, 0x07, 0x00, 0x01, 0x00, 0x00, 0x00]);

        assert_eq!(stream.has_frame().unwrap(), true);
        assert_eq!(
            stream.parse_command().unwrap(),
            Some(ReplayCommand::Advance { ticks: 1 })
        );
        assert_eq!(stream.has_frame().unwrap(), false);
    }

    #[test]
    fn test_stream_feed_reader() {
        let mut stream = StreamParser::new();
        let mut data: &[u8] = &[0x0, 0x07];

        // Feed from one type of `Read`
        assert_eq!(stream.feed_reader(&mut data, 10).unwrap(), 2);

        // Feed from another reader
        let mut cur = Cursor::new(vec![0x00, 0x01, 0x00]);
        assert_eq!(stream.feed_reader(&mut cur, 1).unwrap(), 1);
        assert_eq!(stream.feed_reader(&mut cur, 10).unwrap(), 2);

        // Some more bytes from the same reader again
        cur.write(&[0x00, 0x00]).unwrap();
        cur.set_position(cur.position() - 2);
        assert_eq!(stream.feed_reader(&mut cur, 10).unwrap(), 2);

        assert_eq!(
            stream.parse_command().unwrap(),
            Some(ReplayCommand::Advance { ticks: 1 })
        );
    }
}
