liminal-rs 0.16.4

use super::{
    CausalContext, Frame, MessageEnvelope, ProtocolError, SchemaId, decode, encode, encoded_len,
};
use crate::protocol::{
    DELIVER_MESSAGE_ID_FLAG, DELIVER_MESSAGE_ID_VERSION, DeliveredId, PUBLISH_ACK_POSITION_VERSION,
    ProtocolVersion,
};

/// The frame `Frame::Deliver { flags: 0, stream_id: 9, delivery_seq: 3 }`
/// carrying [`envelope`], as the encoder wrote it before Deliver carried a
/// message id (protocol 1.0 and 1.1): the 10-byte header (type 0x19, flags 0,
/// stream 9, payload length 56), `delivery_seq` as a big-endian `u64`, then the
/// envelope as a length-prefixed field (44 bytes: the schema id, the causal
/// context's length and its two absent tags, the payload's length and bytes).
const OLDER_DELIVER: [u8; 66] = [
    0x19, 0x00, 0x00, 0x00, 0x00, 0x09, 0x00, 0x00, 0x00, 0x38, // header
    0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03, // delivery_seq
    0x00, 0x00, 0x00, 0x2C, // envelope length
    0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11,
    0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11, 0x11,
    0x00, 0x00, 0x00, 0x02, 0x00, 0x00, // causal context: no parent, no clock
    0x00, 0x00, 0x00, 0x02, 0xAA, 0xBB, // payload
];

fn envelope() -> MessageEnvelope {
    MessageEnvelope::new(
        SchemaId::new([0x11; SchemaId::WIRE_LEN]),
        CausalContext::independent(),
        vec![0xAA, 0xBB],
    )
}

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)
}

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

/// The boot and id the newer frame carries.
const ID: DeliveredId = DeliveredId {
    boot: 1_791_328_512_345,
    id: 42,
};

fn deliver(message: Option<DeliveredId>) -> Frame {
    Frame::Deliver {
        flags: 0,
        stream_id: 9,
        delivery_seq: 3,
        envelope: envelope(),
        message,
    }
}

/// The golden a 1.0 or 1.1 peer is owed: a Deliver with no message id is
/// byte-equal to the frame before Deliver could carry one.
#[test]
fn deliver_without_a_message_id_keeps_the_older_layout() -> Result<(), ProtocolError> {
    assert_eq!(encode_frame(&deliver(None))?, OLDER_DELIVER.to_vec());
    let (frame, consumed) = decode(&OLDER_DELIVER)?;
    assert_eq!(consumed, OLDER_DELIVER.len());
    assert_eq!(frame, deliver(None));
    Ok(())
}

/// 4c (Hermes 08:07, 08:10): the message id rides at protocol 1.2, after the
/// publish ack's durable position at 1.1.
#[test]
fn the_message_id_is_carried_from_protocol_1_2() {
    assert_eq!(DELIVER_MESSAGE_ID_VERSION, ProtocolVersion::new(1, 2));
    assert!(DELIVER_MESSAGE_ID_VERSION > PUBLISH_ACK_POSITION_VERSION);
}

/// A Deliver carrying an id appends the boot, then the id, each a big-endian
/// `u64`, after the envelope, and sets the bit; nothing before them moves.
#[test]
fn deliver_with_a_message_id_appends_the_boot_then_the_id() -> Result<(), ProtocolError> {
    let bytes = encode_frame(&deliver(Some(ID)))?;
    assert_eq!(bytes.len(), OLDER_DELIVER.len() + 16);
    assert_eq!(bytes.get(FLAGS_BYTE), Some(&DELIVER_MESSAGE_ID_FLAG));
    assert_eq!(bytes.get(9), Some(&0x48), "the payload length grows by 16");
    assert_eq!(bytes.get(10..OLDER_DELIVER.len()), OLDER_DELIVER.get(10..));
    let mut tail = ID.boot.to_be_bytes().to_vec();
    tail.extend_from_slice(&ID.id.to_be_bytes());
    assert_eq!(bytes.get(OLDER_DELIVER.len()..), Some(tail.as_slice()));
    Ok(())
}

#[test]
fn newer_deliver_decodes_with_its_message_id() -> Result<(), ProtocolError> {
    let bytes = encode_frame(&deliver(Some(ID)))?;
    let (frame, consumed) = decode(&bytes)?;
    assert_eq!(consumed, bytes.len());
    assert_eq!(frame.flags(), DELIVER_MESSAGE_ID_FLAG);
    assert!(matches!(
        frame,
        Frame::Deliver {
            message: Some(ID),
            ..
        }
    ));
    Ok(())
}

/// The bit is the codec's to set: it follows the field, so a frame whose
/// flags claim an id it does not carry is written without the bit.
#[test]
fn the_bit_follows_the_field() -> Result<(), ProtocolError> {
    let claimed = Frame::Deliver {
        flags: DELIVER_MESSAGE_ID_FLAG,
        stream_id: 9,
        delivery_seq: 3,
        envelope: envelope(),
        message: None,
    };
    assert_eq!(encode_frame(&claimed)?, OLDER_DELIVER.to_vec());
    Ok(())
}

#[test]
fn deliver_flag_without_its_message_id_is_refused() {
    let mut bytes = OLDER_DELIVER;
    bytes[FLAGS_BYTE] = DELIVER_MESSAGE_ID_FLAG;
    assert!(matches!(
        decode(&bytes),
        Err(ProtocolError::CodecError { .. })
    ));
}

/// Half a pair (a boot with no id) is refused, never read as an id of 0.
#[test]
fn deliver_with_half_its_message_id_is_refused() -> Result<(), ProtocolError> {
    let mut bytes = encode_frame(&deliver(Some(ID)))?;
    bytes.truncate(bytes.len() - 8);
    if let Some(length) = bytes.get_mut(9) {
        *length = 0x40;
    }
    assert!(matches!(
        decode(&bytes),
        Err(ProtocolError::CodecError { .. })
    ));
    Ok(())
}

/// 4c (Hermes 08:10): a reply names its request as `<boot>:<id>`, the tap id's
/// number with the boot in place of the device, so a tap pairs it only within
/// the boot that drew the id.
#[test]
fn a_delivered_id_names_its_reply_parent_as_boot_colon_id() {
    let parent = ID.reply_parent();
    assert_eq!(parent, super::MessageId::from("1791328512345:42"));
    assert_eq!(DeliveredId::from_parent(&parent), Some(ID));
}

/// Only the form [`DeliveredId::reply_parent`] writes is read back: anything
/// else names no request, so a tap never pairs on a guess.
#[test]
fn a_parent_of_another_form_names_no_request() {
    for other in [
        "local:42",
        "42",
        "1:2:3",
        "",
        "1:",
        ":2",
        "01:2",
        "1:02",
        "+1:2",
        "1:+2",
        "1: 2",
        "-1:2",
        "18446744073709551616:2",
    ] {
        assert_eq!(
            DeliveredId::from_parent(&super::MessageId::from(other)),
            None,
            "{other:?}"
        );
    }
    assert_eq!(
        DeliveredId::from_parent(&super::MessageId::from("0:0")),
        Some(DeliveredId { boot: 0, id: 0 })
    );
}