liminal-rs 0.16.4

use super::{Frame, ProtocolError, decode, encode, encoded_len};

/// The ack `Frame::PublishAck { flags: 0x01, stream_id: 1, message_id: 7 }` as
/// the source commit's encoder wrote it, before the durable position existed:
/// the 10-byte header (type 0x0A, flags 0x01, stream 1, payload length 8)
/// followed by `message_id` as a big-endian `u64`.
const OLDER_ACK: [u8; 18] = [
    0x0A, 0x01, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
    0x00, 0x07,
];

/// Index of the header's flags byte, where the durable-position bit rides.
const FLAGS_BYTE: usize = 1;

fn encode_frame(frame: &Frame) -> Result<Vec<u8>, ProtocolError> {
    let mut bytes = vec![0_u8; encoded_len(frame)?];
    let written = encode(frame, &mut bytes)?;
    bytes.truncate(written);
    Ok(bytes)
}

fn ack(durable_position: Option<u64>) -> Frame {
    Frame::PublishAck {
        flags: 0x01,
        stream_id: 1,
        message_id: 7,
        durable_position,
    }
}

#[test]
fn publish_ack_without_position_keeps_the_older_layout() -> Result<(), ProtocolError> {
    assert_eq!(encode_frame(&ack(None))?, OLDER_ACK.to_vec());
    Ok(())
}

#[test]
fn publish_ack_with_position_appends_it() -> Result<(), ProtocolError> {
    let bytes = encode_frame(&ack(Some(7717)))?;
    assert_eq!(bytes.len(), OLDER_ACK.len() + 8);
    let Some(tail) = bytes.get(bytes.len() - 8..) else {
        return Err(ProtocolError::codec(
            "encoded ack shorter than its position",
        ));
    };
    assert_eq!(tail, 7717_u64.to_be_bytes().as_slice());
    assert_eq!(bytes.get(FLAGS_BYTE), Some(&0x03));
    Ok(())
}

#[test]
fn older_publish_ack_decodes_without_position() -> Result<(), ProtocolError> {
    let (frame, consumed) = decode(&OLDER_ACK)?;
    assert_eq!(consumed, OLDER_ACK.len());
    assert_eq!(frame, ack(None));
    Ok(())
}

#[test]
fn newer_publish_ack_decodes_with_position() -> Result<(), ProtocolError> {
    let bytes = encode_frame(&ack(Some(7717)))?;
    let (frame, consumed) = decode(&bytes)?;
    assert_eq!(consumed, bytes.len());
    assert!(matches!(
        frame,
        Frame::PublishAck {
            stream_id: 1,
            message_id: 7,
            durable_position: Some(7717),
            ..
        }
    ));
    Ok(())
}

#[test]
fn publish_ack_flag_without_position_is_refused() {
    let mut bytes = OLDER_ACK;
    bytes[FLAGS_BYTE] = 0x03;
    assert!(matches!(
        decode(&bytes),
        Err(ProtocolError::CodecError { .. })
    ));
}