use bytes::Bytes;
use serde::Deserialize;
use tailsurf::{
ClientWriterId, WriterId,
protocol::{
rest::{StreamMetadata, Visibility},
ws::{
MAX_WRITER_IN_FLIGHT_PAYLOAD_BYTES, MAX_WRITER_IN_FLIGHT_RECORDS,
WEBSOCKET_HEARTBEAT_INTERVAL_MS,
frame::{
AppendRecord, CaughtUpPosition, ClientFrame, MAX_APPEND_FRAME_RECORDS,
MAX_ENCODED_FRAME_BYTES, MAX_FRAME_PAYLOAD_BYTES, MAX_READ_FRAME_RECORDS,
MAX_RECORD_PAYLOAD_BYTES, OwnedReadRecord, PartHeader, ReadBatch, RecordFormat,
ServerFrame, TSF_WEBSOCKET_PROTOCOL,
},
},
},
};
const FIXTURES_JSON: &str = include_str!("../fixtures/v1.json");
#[derive(Deserialize)]
struct Fixtures {
websocket_protocol: String,
websocket_heartbeat_interval_ms: u64,
max_record_payload_bytes: usize,
max_append_frame_records: usize,
max_read_frame_records: usize,
max_frame_payload_bytes: usize,
max_encoded_frame_bytes: usize,
max_writer_in_flight_records: usize,
max_writer_in_flight_payload_bytes: usize,
client_frames: Vec<FrameFixture<ClientFixture>>,
server_frames: Vec<FrameFixture<ServerFixture>>,
}
#[derive(Deserialize)]
struct FrameFixture<T> {
name: String,
frame: T,
hex: String,
}
#[derive(Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum ClientFixture {
OpenRead {
link_secret: Option<String>,
},
OpenWrite {
client_writer_id_hex: String,
link_secret: String,
expected_next_seq_num: Option<String>,
},
AppendBatch {
writer_seq_num: String,
part_raw: String,
format: u8,
data_hex: String,
},
}
#[derive(Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum ServerFixture {
Ready,
AppendAck {
writer_start_seq_num: String,
writer_end_seq_num: String,
start_seq_num: String,
end_seq_num: String,
},
ReadBatch {
seq_num: String,
timestamp_ms: String,
writer_id_hex: String,
writer_seq_num: String,
part_raw: String,
format: u8,
data_hex: String,
},
Heartbeat,
CaughtUp {
next_seq_num: String,
last_timestamp_ms: String,
},
StreamMetadata {
stream_id: String,
title: Option<String>,
visibility: Visibility,
created_at: String,
expires_at: String,
},
}
#[test]
fn protocol_constants_match_v1_fixtures() {
let fixtures = fixtures();
assert_eq!(fixtures.websocket_protocol, TSF_WEBSOCKET_PROTOCOL);
assert_eq!(
fixtures.websocket_heartbeat_interval_ms,
WEBSOCKET_HEARTBEAT_INTERVAL_MS
);
assert_eq!(fixtures.max_record_payload_bytes, MAX_RECORD_PAYLOAD_BYTES);
assert_eq!(fixtures.max_append_frame_records, MAX_APPEND_FRAME_RECORDS);
assert_eq!(fixtures.max_read_frame_records, MAX_READ_FRAME_RECORDS);
assert_eq!(fixtures.max_frame_payload_bytes, MAX_FRAME_PAYLOAD_BYTES);
assert_eq!(fixtures.max_encoded_frame_bytes, MAX_ENCODED_FRAME_BYTES);
assert_eq!(
fixtures.max_writer_in_flight_records,
MAX_WRITER_IN_FLIGHT_RECORDS
);
assert_eq!(
fixtures.max_writer_in_flight_payload_bytes,
MAX_WRITER_IN_FLIGHT_PAYLOAD_BYTES
);
}
#[test]
fn client_frames_match_v1_fixtures() {
let fixtures = fixtures();
for fixture in fixtures.client_frames {
let expected = decode_hex(&fixture.hex);
let encoded = client_frame(fixture.frame)
.encode()
.unwrap_or_else(|error| panic!("{} fixture failed to encode: {error}", fixture.name));
assert_eq!(encoded.as_ref(), expected, "{} fixture", fixture.name);
let decoded = ClientFrame::decode(&expected)
.unwrap_or_else(|error| panic!("{} fixture failed to decode: {error}", fixture.name));
let reencoded = decoded.encode().unwrap_or_else(|error| {
panic!("{} fixture failed to re-encode: {error}", fixture.name)
});
assert_eq!(reencoded.as_ref(), expected, "{} fixture", fixture.name);
}
}
#[test]
fn server_frames_match_v1_fixtures() {
let fixtures = fixtures();
for fixture in fixtures.server_frames {
let expected = decode_hex(&fixture.hex);
let frame = server_frame(fixture.frame);
let encoded = frame
.encode()
.unwrap_or_else(|error| panic!("{} fixture failed to encode: {error}", fixture.name));
assert_eq!(encoded.as_ref(), expected, "{} fixture", fixture.name);
let decoded = ServerFrame::decode(&expected)
.unwrap_or_else(|error| panic!("{} fixture failed to decode: {error}", fixture.name));
assert_eq!(decoded, frame, "{} fixture", fixture.name);
}
}
fn fixtures() -> Fixtures {
serde_json::from_str(FIXTURES_JSON).expect("v1 protocol fixtures are valid JSON")
}
fn client_frame(fixture: ClientFixture) -> ClientFrame {
match fixture {
ClientFixture::OpenRead { link_secret } => ClientFrame::OpenRead {
link_secret: link_secret
.map(|secret| secret.parse().expect("fixture link secret is canonical")),
},
ClientFixture::OpenWrite {
client_writer_id_hex,
link_secret,
expected_next_seq_num,
} => ClientFrame::OpenWrite {
client_writer_id: decode_client_writer_id(&client_writer_id_hex),
link_secret: link_secret
.parse()
.expect("fixture link secret is canonical"),
expected_next_seq_num: expected_next_seq_num.as_deref().map(parse_u64),
},
ClientFixture::AppendBatch {
writer_seq_num,
part_raw,
format,
data_hex,
} => ClientFrame::AppendBatch(vec![AppendRecord {
writer_seq_num: parse_u64(&writer_seq_num),
part: PartHeader::from_raw(parse_hex_u32(&part_raw)),
format: parse_format(format),
data: Bytes::from(decode_hex(&data_hex)),
}]),
}
}
fn server_frame(fixture: ServerFixture) -> ServerFrame {
match fixture {
ServerFixture::Ready => ServerFrame::Ready,
ServerFixture::AppendAck {
writer_start_seq_num,
writer_end_seq_num,
start_seq_num,
end_seq_num,
} => ServerFrame::AppendAck {
writer_start_seq_num: parse_u64(&writer_start_seq_num),
writer_end_seq_num: parse_u64(&writer_end_seq_num),
start_seq_num: parse_u64(&start_seq_num),
end_seq_num: parse_u64(&end_seq_num),
},
ServerFixture::ReadBatch {
seq_num,
timestamp_ms,
writer_id_hex,
writer_seq_num,
part_raw,
format,
data_hex,
} => ServerFrame::ReadBatch(
ReadBatch::try_from_records(vec![OwnedReadRecord {
seq_num: parse_u64(&seq_num),
timestamp_ms: parse_u64(×tamp_ms),
writer_id: decode_writer_id(&writer_id_hex),
writer_seq_num: parse_u64(&writer_seq_num),
part: PartHeader::from_raw(parse_hex_u32(&part_raw)),
format: parse_format(format),
data: Bytes::from(decode_hex(&data_hex)),
}])
.expect("fixture record within batch bounds"),
),
ServerFixture::Heartbeat => ServerFrame::Heartbeat,
ServerFixture::CaughtUp {
next_seq_num,
last_timestamp_ms,
} => ServerFrame::CaughtUp(CaughtUpPosition {
next_seq_num: parse_u64(&next_seq_num),
last_timestamp_ms: parse_u64(&last_timestamp_ms),
}),
ServerFixture::StreamMetadata {
stream_id,
title,
visibility,
created_at,
expires_at,
} => ServerFrame::StreamMetadata(StreamMetadata {
stream_id: stream_id.parse().expect("fixture stream ID"),
title: title.map(|title| title.parse().expect("fixture stream title")),
visibility,
created_at,
expires_at,
}),
}
}
fn parse_u64(value: &str) -> u64 {
value.parse().expect("fixture value is a u64")
}
fn parse_hex_u32(value: &str) -> u32 {
u32::from_str_radix(value, 16).expect("fixture value is a hexadecimal u32")
}
fn parse_format(value: u8) -> RecordFormat {
RecordFormat::try_from(value).expect("fixture record format is valid")
}
fn decode_writer_id(value: &str) -> WriterId {
let bytes: [u8; WriterId::BYTE_LEN] = decode_hex(value)
.try_into()
.expect("fixture writer ID has the correct length");
WriterId::from_bytes(bytes)
}
fn decode_client_writer_id(value: &str) -> ClientWriterId {
let bytes: [u8; ClientWriterId::BYTE_LEN] = decode_hex(value)
.try_into()
.expect("fixture client writer ID has the correct length");
ClientWriterId::from_bytes(bytes)
}
fn decode_hex(value: &str) -> Vec<u8> {
assert!(
value.len().is_multiple_of(2),
"fixture hex has an even length"
);
value
.as_bytes()
.as_chunks::<2>()
.0
.iter()
.map(|digits| {
let digits = std::str::from_utf8(digits).expect("fixture hex is ASCII");
u8::from_str_radix(digits, 16).expect("fixture hex contains hexadecimal bytes")
})
.collect()
}