liminal-bus-protocol 0.2.0

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,
        boot: None,
    }
}

#[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 { .. })
    ));
}

#[test]
fn publish_ack_boot_round_trips_with_and_without_position() -> Result<(), ProtocolError> {
    for position in [None, Some(7717)] {
        let frame = Frame::PublishAck {
            flags: 0x01,
            stream_id: 1,
            message_id: 7,
            durable_position: position,
            boot: Some(1234),
        };
        let bytes = encode_frame(&frame)?;
        let expected_flags = if position.is_some() { 0x07 } else { 0x05 };
        assert_eq!(bytes.get(FLAGS_BYTE), Some(&expected_flags));
        assert_eq!(
            bytes.get(bytes.len() - 8..),
            Some(1234_u64.to_be_bytes().as_slice())
        );
        let (decoded, consumed) = decode(&bytes)?;
        assert_eq!(consumed, bytes.len());
        assert!(matches!(decoded, Frame::PublishAck {
            message_id: 7, boot: Some(1234), durable_position, ..
        } if durable_position == position));
    }
    Ok(())
}

#[test]
fn publish_ack_short_boot_is_refused_by_name() -> Result<(), ProtocolError> {
    for position in [None, Some(7717)] {
        for tail_length in 0_u8..8 {
            let mut bytes = encode_frame(&ack(position))?;
            bytes[FLAGS_BYTE] |= 0x04;
            // Both legacy bodies fit in one length byte in the header.
            bytes[9] += tail_length;
            bytes.resize(bytes.len() + usize::from(tail_length), 0);
            assert!(
                matches!(decode(&bytes), Err(ProtocolError::CodecError { message })
                if message.as_deref() == Some("publish_ack_boot_truncated"))
            );
        }
    }
    Ok(())
}

#[test]
fn publish_ack_without_boot_keeps_position_layout_exactly() -> Result<(), ProtocolError> {
    assert_eq!(
        encode_frame(&ack(Some(7717)))?,
        vec![
            0x0A, 0x03, 0, 0, 0, 1, 0, 0, 0, 16, 0, 0, 0, 0, 0, 0, 0, 7, 0, 0, 0, 0, 0, 0, 0x1E,
            0x25,
        ]
    );
    Ok(())
}

#[test]
fn publish_ack_clears_optional_flags_without_fields() -> Result<(), ProtocolError> {
    let frame = Frame::PublishAck {
        flags: 0x07,
        stream_id: 1,
        message_id: 7,
        durable_position: None,
        boot: None,
    };
    assert_eq!(encode_frame(&frame)?, OLDER_ACK);
    Ok(())
}