pi_append_log 0.3.0

//! 追加日志默认块格式的公共常量、编解码接口和已解析数据块视图。

use std::io;

use crc32fast::Hasher;
use pi_result::InteropResultExt;

/// 默认 V1 envelope 的 ASCII Magic 字节序列。
pub const DEFAULT_MAGIC_BYTES: [u8; 4] = *b"pial";

/// 默认 V1 envelope 按小端序解释后的 Magic 整数值。
pub const DEFAULT_MAGIC: u32 = u32::from_le_bytes(DEFAULT_MAGIC_BYTES);

/// 默认 V1 envelope 的格式版本号。
pub const DEFAULT_VERSION: u16 = 1;

/// 默认 V1 envelope 在 Payload 之前、从 Magic 到 BlockSeq 结束的固定字节数。
pub const DEFAULT_BODY_FIXED_LEN: usize = 16;

/// 默认 V1 envelope 除 Payload 外的总固定字节数。
///
/// 该值包含前置 BodyLen、固定 Body、后置 BodyLen 和 CRC32。
pub const DEFAULT_ENCODED_FIXED_LEN: usize = 28;

/// 默认 DefaultBlockCodec 使用的最大 Payload 字节数，即 4 MiB。
pub const DEFAULT_MAX_PAYLOAD_LEN: usize = 4 * 1024 * 1024;

/// 默认 V1 block codec 的配置。
///
/// 此类型只保存由调用方传入的最大 Payload 限制。默认值为 4 MiB；具体 codec 实现
/// 必须在编码、解码和尾部探测中一致执行该限制。
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct DefaultBlockCodec {
    /// 单个 Payload 允许的最大字节数。
    max_payload_len: usize,
}

impl DefaultBlockCodec {
    /// 使用调用方指定的最大 Payload 字节数创建默认 V1 codec 配置。
    ///
    /// 上限不得超过 V1 的 u32 BodyLen 可表示范围；具体 codec 实现必须在创建或编码
    /// 时拒绝无法表示的值。
    pub fn new(max_payload_len: usize) -> Self {
        Self { max_payload_len }
    }

    /// 返回调用方配置的最大 Payload 字节数。
    pub fn max_payload_len(&self) -> usize {
        self.max_payload_len
    }
}

impl Default for DefaultBlockCodec {
    /// 使用 4 MiB 的默认最大 Payload 限制创建 codec 配置。
    fn default() -> Self {
        Self::new(DEFAULT_MAX_PAYLOAD_LEN)
    }
}

impl BlockEncoder for DefaultBlockCodec {
    type Block = Vec<u8>;

    fn encode(&self, block_seq: u64, flags: u16, payload: &[u8]) -> pi_result::Result<Self::Block> {
        if block_seq == 0 {
            return Err(invalid_input("block sequence must be greater than zero"));
        }
        if flags != 0 {
            return Err(invalid_input("V1 does not support non-zero flags"));
        }
        validate_payload_len(self.max_payload_len, payload.len())?;
        let body_len = DEFAULT_BODY_FIXED_LEN
            .checked_add(payload.len())
            .ok_or_else(|| invalid_input("body length overflow"))?;
        let body_len_u32 = u32::try_from(body_len)
            .map_err(|_| invalid_input("body length does not fit in u32"))?;
        let encoded_len = body_len
            .checked_add(12)
            .ok_or_else(|| invalid_input("encoded length overflow"))?;
        let mut encoded = Vec::with_capacity(encoded_len);
        encoded.extend_from_slice(&body_len_u32.to_le_bytes());
        encoded.extend_from_slice(&DEFAULT_MAGIC_BYTES);
        encoded.extend_from_slice(&DEFAULT_VERSION.to_le_bytes());
        encoded.extend_from_slice(&flags.to_le_bytes());
        encoded.extend_from_slice(&block_seq.to_le_bytes());
        encoded.extend_from_slice(payload);
        encoded.extend_from_slice(&body_len_u32.to_le_bytes());
        encoded.extend_from_slice(&crc32(&encoded).to_le_bytes());
        Ok(encoded)
    }
}

impl BlockDecoder for DefaultBlockCodec {
    type Block = Vec<u8>;

    fn decode_forward<'a>(&self, input: &'a [u8]) -> pi_result::Result<DecodedBlock<'a>> {
        if input.len() < DEFAULT_ENCODED_FIXED_LEN {
            return Err(invalid_data("input is shorter than the V1 envelope"));
        }
        let body_len = read_u32(input, 0)? as usize;
        let encoded_len = checked_encoded_len(body_len)?;
        if encoded_len > input.len() {
            return Err(invalid_data("input does not contain a complete block"));
        }
        decode_exact(self.max_payload_len, &input[..encoded_len])
    }

    fn decode_backward<'a>(&self, input: &'a [u8]) -> pi_result::Result<DecodedBlock<'a>> {
        if input.len() < DEFAULT_ENCODED_FIXED_LEN {
            return Err(invalid_data("input is shorter than the V1 envelope"));
        }
        let trailer_start = input.len() - 8;
        let body_len = read_u32(input, trailer_start)? as usize;
        let encoded_len = checked_encoded_len(body_len)?;
        if encoded_len > input.len() {
            return Err(invalid_data("input does not contain a complete block"));
        }
        decode_exact(self.max_payload_len, &input[input.len() - encoded_len..])
    }

    fn find_last_complete(&self, input: &[u8]) -> pi_result::Result<Option<usize>> {
        if input.len() < DEFAULT_ENCODED_FIXED_LEN {
            return Ok(None);
        }
        let mut search_end = input.len();
        while search_end >= DEFAULT_MAGIC_BYTES.len() {
            let Some(relative_magic) = input[..search_end]
                .windows(DEFAULT_MAGIC_BYTES.len())
                .rposition(|window| window == DEFAULT_MAGIC_BYTES)
            else {
                break;
            };
            let magic_start = relative_magic;
            let Some(prefix_start) = magic_start.checked_sub(4) else {
                search_end = magic_start;
                continue;
            };
            let Ok(body_len) = read_u32(input, prefix_start).map(|length| length as usize) else {
                search_end = magic_start;
                continue;
            };
            let Ok(encoded_len) = checked_encoded_len(body_len) else {
                search_end = magic_start;
                continue;
            };
            let Some(candidate_end) = prefix_start.checked_add(encoded_len) else {
                search_end = magic_start;
                continue;
            };
            if candidate_end <= input.len()
                && decode_exact(self.max_payload_len, &input[prefix_start..candidate_end]).is_ok()
            {
                return Ok(Some(candidate_end));
            }
            search_end = magic_start;
        }
        Ok(None)
    }
}

fn checked_encoded_len(body_len: usize) -> pi_result::Result<usize> {
    if body_len < DEFAULT_BODY_FIXED_LEN {
        return Err(invalid_data(
            "body length is smaller than the V1 fixed body",
        ));
    }
    body_len
        .checked_add(12)
        .ok_or_else(|| invalid_data("encoded length overflow"))
}

fn decode_exact<'a>(
    max_payload_len: usize,
    input: &'a [u8],
) -> pi_result::Result<DecodedBlock<'a>> {
    let body_len = read_u32(input, 0)? as usize;
    let encoded_len = checked_encoded_len(body_len)?;
    if encoded_len != input.len() {
        return Err(invalid_data("encoded length does not match input"));
    }
    let payload_len = body_len - DEFAULT_BODY_FIXED_LEN;
    validate_payload_len(max_payload_len, payload_len)
        .map_err(|error| invalid_data(&error.to_string()))?;
    if input[4..8] != DEFAULT_MAGIC_BYTES {
        return Err(invalid_data("invalid V1 magic"));
    }
    if read_u16(input, 8)? != DEFAULT_VERSION {
        return Err(invalid_data("unsupported V1 version"));
    }
    let flags = read_u16(input, 10)?;
    if flags != 0 {
        return Err(invalid_data("V1 does not support non-zero flags"));
    }
    let block_seq = read_u64(input, 12)?;
    if block_seq == 0 {
        return Err(invalid_data("block sequence must be greater than zero"));
    }
    let suffix_start = 20 + payload_len;
    if read_u32(input, suffix_start)? as usize != body_len {
        return Err(invalid_data("prefix and suffix body lengths differ"));
    }
    let stored_crc = read_u32(input, suffix_start + 4)?;
    if stored_crc != crc32(&input[..input.len() - 4]) {
        return Err(invalid_data("CRC32 validation failed"));
    }
    Ok(DecodedBlock {
        encoded: input,
        payload: &input[20..suffix_start],
        block_seq,
        flags,
    })
}

fn read_u16(input: &[u8], offset: usize) -> pi_result::Result<u16> {
    let bytes = input
        .get(offset..offset + 2)
        .ok_or_else(|| invalid_data("truncated u16"))?;
    Ok(u16::from_le_bytes([bytes[0], bytes[1]]))
}

fn read_u32(input: &[u8], offset: usize) -> pi_result::Result<u32> {
    let bytes = input
        .get(offset..offset + 4)
        .ok_or_else(|| invalid_data("truncated u32"))?;
    Ok(u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]))
}

fn read_u64(input: &[u8], offset: usize) -> pi_result::Result<u64> {
    let bytes = input
        .get(offset..offset + 8)
        .ok_or_else(|| invalid_data("truncated u64"))?;
    let array: [u8; 8] = bytes
        .try_into()
        .map_err(|_| invalid_data("truncated u64"))?;
    Ok(u64::from_le_bytes(array))
}

fn invalid_data(message: &str) -> pi_result::Error {
    Err::<(), _>(io::Error::new(io::ErrorKind::InvalidData, message))
        .into_classified_error()
        .expect_err("invalid data conversion must fail")
}

fn validate_payload_len(max_payload_len: usize, payload_len: usize) -> pi_result::Result<()> {
    if payload_len > max_payload_len {
        return Err(invalid_input("payload exceeds configured maximum"));
    }
    if payload_len > (u32::MAX as usize).saturating_sub(DEFAULT_BODY_FIXED_LEN) {
        return Err(invalid_input("payload cannot be represented by V1"));
    }
    Ok(())
}

fn crc32(bytes: &[u8]) -> u32 {
    let mut hasher = Hasher::new();
    hasher.update(bytes);
    hasher.finalize()
}

fn invalid_input(message: &str) -> pi_result::Error {
    Err::<(), _>(io::Error::new(io::ErrorKind::InvalidInput, message))
        .into_classified_error()
        .expect_err("invalid input conversion must fail")
}

/// 编码完整 envelope block 的接口。
pub trait BlockEncoder: Send + Sync {
    /// 编码后可交给 AppendLog 追加的完整 block 类型。
    type Block: AsRef<[u8]> + Clone + Send + Sync + 'static;

    /// 使用指定序号、Flags 和 Payload 编码一个完整 block。
    ///
    /// block_seq 必须大于零。具体 encoder 必须拒绝不支持的 Flags、超出最大长度的
    /// Payload 以及不能用目标格式表示的字段。
    fn encode(&self, block_seq: u64, flags: u16, payload: &[u8]) -> pi_result::Result<Self::Block>;
}

/// 严格解码和尾部恢复探测完整 envelope block 的接口。
pub trait BlockDecoder: Send + Sync {
    /// 此 decoder 对应的完整编码 block 类型。
    type Block: AsRef<[u8]> + Clone + Send + Sync + 'static;

    /// 从输入起始位置严格解码一个完整 block。
    ///
    /// 输入可以在当前 block 后附带更多字节；调用方通过 DecodedBlock::encoded_len
    /// 确定下一个 block 的起始位置。结构、Magic、版本、Flags、序号或 CRC32 错误都
    /// 必须直接返回错误，不能静默跳过。
    fn decode_forward<'a>(&self, input: &'a [u8]) -> pi_result::Result<DecodedBlock<'a>>;

    /// 从输入末尾的候选结束位置严格反向解码一个完整 block。
    ///
    /// 输入可以在当前 block 前包含更多字节；返回的 DecodedBlock 借用其中位于末尾的
    /// 完整 block。结构、Magic、版本、Flags、序号或 CRC32 错误都必须直接返回错误。
    fn decode_backward<'a>(&self, input: &'a [u8]) -> pi_result::Result<DecodedBlock<'a>>;

    /// 在可能带有半写或垃圾尾部的输入中寻找最近完整 block 的结束偏移。
    ///
    /// 此方法只用于活动结构的尾部恢复。它可以跳过文件尾部连续的半写或垃圾字节，
    /// 但不能把中间损坏静默视为正常数据。没有任何完整 block 时返回 Ok(None)。
    fn find_last_complete(&self, input: &[u8]) -> pi_result::Result<Option<usize>>;
}

/// 借用输入字节的已严格验证 envelope block 视图。
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct DecodedBlock<'a> {
    /// 通过严格校验的完整 envelope block 字节。
    encoded: &'a [u8],
    /// 完整 envelope 中对应业务内容的 Payload 字节。
    payload: &'a [u8],
    /// 从 envelope 解码出的非零逻辑序号。
    block_seq: u64,
    /// 从 envelope 解码出的格式 Flags。
    flags: u16,
}

impl<'a> DecodedBlock<'a> {
    /// 返回包含 envelope、Payload、后置长度和 CRC32 的完整已编码 block。
    pub fn encoded(&self) -> &'a [u8] {
        self.encoded
    }

    /// 返回完整 block 中的原始 Payload 字节。
    pub fn payload(&self) -> &'a [u8] {
        self.payload
    }

    /// 返回当前 block 的非零逻辑序号。
    pub fn block_seq(&self) -> u64 {
        self.block_seq
    }

    /// 返回当前 block 的格式 Flags。
    pub fn flags(&self) -> u16 {
        self.flags
    }

    /// 返回完整已编码 block 的字节长度。
    pub fn encoded_len(&self) -> usize {
        self.encoded.len()
    }
}