path: root/src/protocol.rs

//! Protocol v5: bounded frames and the request/reply/event envelope.
//!
//! Every packet is `[u32 length][UTF-8 payload]`. The length is big-endian and
//! capped, so a peer cannot make the reader allocate without bound. Inside a
//! frame the payload is text:
//!
//! ```text
//! REQ <id> [idem=<token>] <command>
//! RES <id> <body>
//! EVT <seq> <body>
//! ```
//!
//! The command and body grammar is the same text protocol the games already
//! use (`HELLO`, `PLAY_START`, `GAME_RESULT`, ...). The envelope only adds
//! correlation, event numbering and a hard size limit.

use std::fmt;
use std::io::{self, Read, Write};

/// Largest payload a peer may send in one frame. A game message is a few
/// hundred bytes; 64 KiB is generous and still bounds the reader.
pub const MAX_FRAME_LEN: usize = 64 * 1024;

/// How many request ids a connection remembers for idempotent retries.
pub const REPLAY_CACHE_LEN: usize = 256;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FrameError {
    /// The declared payload length exceeds [`MAX_FRAME_LEN`].
    TooLarge(usize),
    /// The payload was not valid UTF-8.
    InvalidUtf8,
}

impl fmt::Display for FrameError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            FrameError::TooLarge(length) => {
                write!(
                    f,
                    "frame of {length} bytes exceeds the {MAX_FRAME_LEN}-byte limit"
                )
            }
            FrameError::InvalidUtf8 => write!(f, "frame payload is not valid UTF-8"),
        }
    }
}

impl std::error::Error for FrameError {}

/// A bounded frame decoder. Feed it bytes from a socket, take whole frames.
#[derive(Debug, Default)]
pub struct FrameBuffer {
    buffer: Vec<u8>,
}

impl FrameBuffer {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn push(&mut self, bytes: &[u8]) {
        self.buffer.extend_from_slice(bytes);
    }

    /// Return the next complete frame, if the buffer holds one.
    ///
    /// A declared length above the limit is reported immediately instead of
    /// waiting for bytes that will never be accepted; the caller should close
    /// the connection.
    pub fn next_frame(&mut self) -> Option<Result<String, FrameError>> {
        if self.buffer.len() < 4 {
            return None;
        }

        let length = u32::from_be_bytes([
            self.buffer[0],
            self.buffer[1],
            self.buffer[2],
            self.buffer[3],
        ]) as usize;

        if length > MAX_FRAME_LEN {
            self.buffer.clear();
            return Some(Err(FrameError::TooLarge(length)));
        }
        if self.buffer.len() < 4 + length {
            return None;
        }

        let payload: Vec<u8> = self.buffer[4..4 + length].to_vec();
        self.buffer.drain(..4 + length);
        match String::from_utf8(payload) {
            Ok(text) => Some(Ok(text)),
            Err(_) => Some(Err(FrameError::InvalidUtf8)),
        }
    }

    /// Read one frame from a blocking reader, preserving leftover bytes.
    /// `None` means EOF with no complete frame buffered.
    pub fn read_frame<R: Read>(&mut self, reader: &mut R) -> io::Result<Option<String>> {
        loop {
            match self.next_frame() {
                Some(Ok(frame)) => return Ok(Some(frame)),
                Some(Err(error)) => return Err(io::Error::new(io::ErrorKind::InvalidData, error)),
                None => {}
            }

            let mut chunk = [0u8; 4096];
            match reader.read(&mut chunk) {
                Ok(0) => return Ok(None),
                Ok(n) => self.push(&chunk[..n]),
                Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
                Err(error) => return Err(error),
            }
        }
    }

    /// True when a partial frame is buffered (used by tests and diagnostics).
    pub fn is_empty(&self) -> bool {
        self.buffer.is_empty()
    }
}

/// Encode one payload as a length-prefixed frame.
pub fn encode_frame(payload: &str) -> Vec<u8> {
    let length = payload.len() as u32;
    let mut bytes = Vec::with_capacity(4 + payload.len());
    bytes.extend_from_slice(&length.to_be_bytes());
    bytes.extend_from_slice(payload.as_bytes());
    bytes
}

/// Write one payload as a length-prefixed frame.
pub fn write_frame<W: Write>(writer: &mut W, payload: &str) -> io::Result<()> {
    writer.write_all(&encode_frame(payload))?;
    writer.flush()
}

/// One client request. `id` is the correlation key; `idem` is optional
/// metadata for logs and a future cross-connection cache.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Request {
    pub id: u64,
    pub idem: Option<String>,
    pub payload: String,
}

/// A frame received by a client.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Packet {
    Reply { id: u64, body: String },
    Event { seq: u64, body: String },
}

pub fn request_line(id: u64, idem: Option<&str>, payload: &str) -> String {
    match idem {
        Some(idem) if !idem.is_empty() => format!("REQ {id} idem={idem} {payload}"),
        _ => format!("REQ {id} {payload}"),
    }
}

pub fn reply_line(id: u64, body: &str) -> String {
    format!("RES {id} {body}")
}

pub fn event_line(seq: u64, body: &str) -> String {
    format!("EVT {seq} {body}")
}

/// Parse a `REQ` payload. Errors are safe to return to the peer.
pub fn parse_request(frame: &str) -> Result<Request, String> {
    let frame = frame.trim_end();
    let rest = frame
        .strip_prefix("REQ ")
        .ok_or_else(|| "Malformed request: expected REQ".to_string())?;

    let (id_token, mut rest) = split_token(rest);
    let id: u64 = id_token
        .parse()
        .map_err(|_| "Malformed request id".to_string())?;

    let mut idem = None;
    let (next, after) = split_token(rest);
    if let Some(value) = next.strip_prefix("idem=") {
        if value.is_empty() {
            return Err("Malformed idempotency key".to_string());
        }
        idem = Some(value.to_string());
        rest = after;
    }

    Ok(Request {
        id,
        idem,
        payload: rest.to_string(),
    })
}

/// Parse a `RES` or `EVT` payload on the client side.
pub fn parse_packet(frame: &str) -> Result<Packet, String> {
    let frame = frame.trim_end();
    if let Some(rest) = frame.strip_prefix("RES ") {
        let (id_token, body) = split_token(rest);
        let id: u64 = id_token
            .parse()
            .map_err(|_| "Malformed reply id".to_string())?;
        return Ok(Packet::Reply {
            id,
            body: body.to_string(),
        });
    }
    if let Some(rest) = frame.strip_prefix("EVT ") {
        let (seq_token, body) = split_token(rest);
        let seq: u64 = seq_token
            .parse()
            .map_err(|_| "Malformed event sequence".to_string())?;
        return Ok(Packet::Event {
            seq,
            body: body.to_string(),
        });
    }
    Err(format!("Unexpected packet: {frame}"))
}

fn split_token(rest: &str) -> (&str, &str) {
    let rest = rest.trim_start();
    match rest.find(' ') {
        Some(index) => (&rest[..index], rest[index + 1..].trim_start()),
        None => (rest, ""),
    }
}

/// A stable, user-facing message for a failed request.
pub fn io_failure_message(error: &io::Error) -> &'static str {
    match error.kind() {
        io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock => "ERROR Request timed out",
        _ => "ERROR Connection lost",
    }
}