use crc32c::{crc32c, crc32c_append};
use std::io::{self, Read, Write};
use thiserror::Error;

pub const MAX_FRAME_BODY_BYTES: usize = 64 * 1024 * 1024;
const U64_SIZE_BYTES: usize = 8;
const PAYLOAD_TYPE_SIZE_BYTES: usize = 1;
const CRC32C_SIZE_BYTES: usize = 4;
const MIN_FRAME_BODY_SIZE_BYTES: usize =
    U64_SIZE_BYTES + U64_SIZE_BYTES + PAYLOAD_TYPE_SIZE_BYTES + CRC32C_SIZE_BYTES;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Frame {
    pub frame_length: u32,
    pub commit_seq: u64,
    pub timestamp_micros: u64,
    pub payload_type: u8,
    pub payload: Vec<u8>,
    pub crc32c: u32,
}

#[derive(Debug, Error, PartialEq, Eq)]
pub enum FrameError {
    #[error("truncated frame")]
    Truncation,
    #[error("corrupt frame")]
    Corruption,
    #[error("io error: {0}")]
    Io(String),
}

impl From<io::Error> for FrameError {
    fn from(value: io::Error) -> Self {
        Self::Io(value.to_string())
    }
}

pub struct FrameWriter<W: Write> {
    inner: W,
}

pub(crate) fn append_frame_bytes(
    out: &mut Vec<u8>,
    commit_seq: u64,
    timestamp_micros: u64,
    payload_type: u8,
    payload: &[u8],
) -> Result<(), FrameError> {
    let frame_body_size_bytes = U64_SIZE_BYTES
        .saturating_add(U64_SIZE_BYTES)
        .saturating_add(PAYLOAD_TYPE_SIZE_BYTES)
        .saturating_add(payload.len())
        .saturating_add(CRC32C_SIZE_BYTES);
    let frame_length = u32::try_from(frame_body_size_bytes).map_err(|_| FrameError::Corruption)?;
    let len_bytes = frame_length.to_be_bytes();
    let seq_bytes = commit_seq.to_be_bytes();
    let ts_bytes = timestamp_micros.to_be_bytes();
    let type_bytes = [payload_type];

    let mut crc = crc32c(&len_bytes);
    crc = crc32c_append(crc, &seq_bytes);
    crc = crc32c_append(crc, &ts_bytes);
    crc = crc32c_append(crc, &type_bytes);
    crc = crc32c_append(crc, payload);
    let crc = crc.to_be_bytes();

    out.reserve(4 + frame_body_size_bytes);
    out.extend_from_slice(&len_bytes);
    out.extend_from_slice(&seq_bytes);
    out.extend_from_slice(&ts_bytes);
    out.extend_from_slice(&type_bytes);
    out.extend_from_slice(payload);
    out.extend_from_slice(&crc);
    Ok(())
}

impl<W: Write> FrameWriter<W> {
    pub fn new(inner: W) -> Self {
        Self { inner }
    }

    pub fn append(
        &mut self,
        commit_seq: u64,
        timestamp_micros: u64,
        payload_type: u8,
        payload: &[u8],
    ) -> Result<(), FrameError> {
        let frame_body_size_bytes = U64_SIZE_BYTES
            .saturating_add(U64_SIZE_BYTES)
            .saturating_add(PAYLOAD_TYPE_SIZE_BYTES)
            .saturating_add(payload.len())
            .saturating_add(CRC32C_SIZE_BYTES);
        let frame_length =
            u32::try_from(frame_body_size_bytes).map_err(|_| FrameError::Corruption)?;
        let len_bytes = frame_length.to_be_bytes();
        let seq_bytes = commit_seq.to_be_bytes();
        let ts_bytes = timestamp_micros.to_be_bytes();
        let type_bytes = [payload_type];

        let mut crc = crc32c(&len_bytes);
        crc = crc32c_append(crc, &seq_bytes);
        crc = crc32c_append(crc, &ts_bytes);
        crc = crc32c_append(crc, &type_bytes);
        crc = crc32c_append(crc, payload);
        let crc = crc.to_be_bytes();

        self.inner.write_all(&len_bytes)?;
        self.inner.write_all(&seq_bytes)?;
        self.inner.write_all(&ts_bytes)?;
        self.inner.write_all(&type_bytes)?;
        self.inner.write_all(payload)?;
        self.inner.write_all(&crc)?;
        Ok(())
    }

    pub fn into_inner(self) -> W {
        self.inner
    }
}

pub struct FrameReader<R: Read> {
    inner: R,
}

impl<R: Read> FrameReader<R> {
    pub fn new(inner: R) -> Self {
        Self { inner }
    }

    pub fn next_frame(&mut self) -> Result<Option<Frame>, FrameError> {
        let mut len_buf = [0u8; 4];
        let first = self.inner.read(&mut len_buf[0..1])?;
        if first == 0 {
            return Ok(None);
        }
        match self.inner.read_exact(&mut len_buf[1..4]) {
            Ok(()) => {}
            Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
                return Err(FrameError::Truncation);
            }
            Err(e) => return Err(FrameError::Io(e.to_string())),
        }
        let frame_length = u32::from_be_bytes(len_buf);
        let frame_body_size_bytes = frame_length as usize;
        if frame_body_size_bytes < MIN_FRAME_BODY_SIZE_BYTES {
            return Err(FrameError::Corruption);
        }
        if frame_body_size_bytes > MAX_FRAME_BODY_BYTES {
            return Err(FrameError::Corruption);
        }

        let mut body = vec![0u8; frame_body_size_bytes];
        match self.inner.read_exact(&mut body) {
            Ok(_) => {}
            Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
                return Err(FrameError::Truncation);
            }
            Err(e) => return Err(FrameError::Io(e.to_string())),
        }

        let crc_offset_bytes = frame_body_size_bytes.saturating_sub(CRC32C_SIZE_BYTES);
        debug_assert!(crc_offset_bytes < frame_body_size_bytes);
        let stored_crc = u32::from_be_bytes(
            body[crc_offset_bytes..]
                .try_into()
                .map_err(|_| FrameError::Corruption)?,
        );
        let mut computed_crc = crc32c(&len_buf);
        computed_crc = crc32c_append(computed_crc, &body[..crc_offset_bytes]);
        if stored_crc != computed_crc {
            return Err(FrameError::Corruption);
        }

        let commit_seq = u64::from_be_bytes(
            body[0..U64_SIZE_BYTES]
                .try_into()
                .map_err(|_| FrameError::Corruption)?,
        );
        let timestamp_micros = u64::from_be_bytes(
            body[U64_SIZE_BYTES..(2 * U64_SIZE_BYTES)]
                .try_into()
                .map_err(|_| FrameError::Corruption)?,
        );
        let payload_type = body[2 * U64_SIZE_BYTES];
        let payload_offset_bytes = (2 * U64_SIZE_BYTES) + PAYLOAD_TYPE_SIZE_BYTES;
        debug_assert!(payload_offset_bytes <= crc_offset_bytes);
        let payload = body[payload_offset_bytes..crc_offset_bytes].to_vec();

        Ok(Some(Frame {
            frame_length,
            commit_seq,
            timestamp_micros,
            payload_type,
            payload,
            crc32c: stored_crc,
        }))
    }
}

#[cfg(test)]
mod tests;
