use super::FirehoseError;
use super::types::{Info, SubscribeReposMessage};
use serde::{Deserialize, Serialize};
#[derive(Debug, Deserialize, Serialize)]
struct EventHeader {
op: i64,
t: Option<String>,
}
#[derive(Deserialize)]
struct ErrorFrame {
error: String,
message: Option<String>,
}
pub fn decode_frame<'i>(bytes: &'i [u8]) -> Result<SubscribeReposMessage<'i>, FirehoseError> {
let mut de = serde_ipld_dagcbor::de::Deserializer::from_slice(bytes);
let header = EventHeader::deserialize(&mut de)?;
match header.op {
-1 => {
let err = ErrorFrame::deserialize(&mut de)?;
return match err.error.as_str() {
"FutureCursor" => Err(FirehoseError::FutureCursor {
message: err.message,
}),
_ => Err(FirehoseError::RelayError {
error: err.error,
message: err.message,
}),
};
}
1 => {}
op => return Err(FirehoseError::UnknownOp(op)),
}
let t = header.t.ok_or(FirehoseError::MissingType)?;
let msg = match t.as_str() {
"#commit" => SubscribeReposMessage::Commit(Box::new(Deserialize::deserialize(&mut de)?)),
"#account" => SubscribeReposMessage::Account(Box::new(Deserialize::deserialize(&mut de)?)),
"#identity" => {
SubscribeReposMessage::Identity(Box::new(Deserialize::deserialize(&mut de)?))
}
"#sync" => SubscribeReposMessage::Sync(Box::new(Deserialize::deserialize(&mut de)?)),
"#info" => {
let info: Info<'i> = Deserialize::deserialize(&mut de)?;
SubscribeReposMessage::Info(Box::new(info))
}
other => return Err(FirehoseError::UnknownType(other.to_string())),
};
Ok(msg)
}
#[cfg(feature = "relay")]
#[derive(Serialize)]
struct EncodeHeader<'a> {
op: i64,
t: &'a str,
}
#[cfg(feature = "relay")]
#[derive(Serialize)]
struct EncodeErrorHeader {
op: i64,
}
#[cfg(feature = "relay")]
#[derive(Serialize)]
struct EncodeErrorFrame<'a> {
error: &'a str,
#[serde(skip_serializing_if = "Option::is_none")]
message: Option<&'a str>,
}
#[cfg(feature = "relay")]
pub fn encode_frame<T: serde::Serialize>(t: &str, msg: &T) -> miette::Result<bytes::Bytes> {
let mut buf = serde_ipld_dagcbor::to_vec(&EncodeHeader { op: 1, t })
.map_err(|e| miette::miette!("encode_frame header: {e}"))?;
buf.extend_from_slice(
&serde_ipld_dagcbor::to_vec(msg).map_err(|e| miette::miette!("encode_frame body: {e}"))?,
);
Ok(bytes::Bytes::from(buf))
}
#[cfg(feature = "relay")]
pub fn encode_error_frame(error: &str, message: Option<&str>) -> miette::Result<bytes::Bytes> {
let mut buf = serde_ipld_dagcbor::to_vec(&EncodeErrorHeader { op: -1 })
.map_err(|e| miette::miette!("encode_error_frame header: {e}"))?;
buf.extend_from_slice(
&serde_ipld_dagcbor::to_vec(&EncodeErrorFrame { error, message })
.map_err(|e| miette::miette!("encode_error_frame body: {e}"))?,
);
Ok(bytes::Bytes::from(buf))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ingest::stream::{Datetime, Identity, InfoName, Sync};
use bytes::Bytes;
use jacquard_common::{CowStr, types::string::Did};
fn frame<T: Serialize>(header: EventHeader, body: &T) -> Vec<u8> {
let mut bytes = serde_ipld_dagcbor::to_vec(&header).unwrap();
bytes.extend_from_slice(&serde_ipld_dagcbor::to_vec(body).unwrap());
bytes
}
#[test]
fn firehose_frame_fixture_decodes_identity() {
let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
let body = Identity {
did,
handle: None,
seq: 42,
time: serde_json::from_str::<Datetime>(r#""2026-08-10T12:34:56Z""#).unwrap(),
};
let bytes = frame(
EventHeader {
op: 1,
t: Some("#identity".to_owned()),
},
&body,
);
assert_eq!(
crate::util::hex::encode(&bytes),
"a2617469236964656e74697479626f7001a36364696478206469643a706c633a65777669376e787a796f756e367a687872687336346f697a63736571182a6474696d6574323032362d30382d31305431323a33343a35365a"
);
let decoded = decode_frame(&bytes).unwrap();
let SubscribeReposMessage::Identity(identity) = decoded else {
panic!("expected identity");
};
assert_eq!(identity.did.as_str(), "did:plc:ewvi7nxzyoun6zhxrhs64oiz");
assert_eq!(identity.seq, 42);
assert_eq!(identity.time.0.to_rfc3339(), "2026-08-10T12:34:56+00:00");
assert!(identity.handle.is_none());
}
#[test]
fn firehose_frame_mock_decodes_sync_binary_blocks() {
let bytes = frame(
EventHeader {
op: 1,
t: Some("#sync".to_owned()),
},
&Sync {
blocks: Bytes::from_static(b"stinkpot"),
did: Did::new_static("did:web:example.com").unwrap(),
rev: CowStr::Borrowed("3jzfcijpj2z2a"),
seq: 42,
time: serde_json::from_str::<Datetime>(r#""2026-08-10T12:34:56Z""#).unwrap(),
},
);
assert_eq!(
crate::util::hex::encode(&bytes),
"a26174652373796e63626f7001a563646964736469643a7765623a6578616d706c652e636f6d637265766d336a7a6663696a706a327a326163736571182a6474696d6574323032362d30382d31305431323a33343a35365a66626c6f636b73487374696e6b706f74"
);
let SubscribeReposMessage::Sync(sync) = decode_frame(&bytes).unwrap() else {
panic!("expected sync");
};
assert_eq!(sync.blocks.as_ref(), b"stinkpot");
assert_eq!(sync.did.as_str(), "did:web:example.com");
assert_eq!(sync.rev.as_ref(), "3jzfcijpj2z2a");
assert_eq!(sync.seq, 42);
assert_eq!(sync.time.0.to_rfc3339(), "2026-08-10T12:34:56+00:00");
}
#[test]
fn firehose_frame_rejects_header_and_relay_errors() {
let unknown_op = serde_ipld_dagcbor::to_vec(&EventHeader {
op: 7,
t: Some("#identity".to_owned()),
})
.unwrap();
assert!(matches!(
decode_frame(&unknown_op),
Err(FirehoseError::UnknownOp(7))
));
let missing_type = serde_ipld_dagcbor::to_vec(&EventHeader { op: 1, t: None }).unwrap();
assert!(matches!(
decode_frame(&missing_type),
Err(FirehoseError::MissingType)
));
let unknown_type = serde_ipld_dagcbor::to_vec(&EventHeader {
op: 1,
t: Some("#unknown".to_owned()),
})
.unwrap();
assert!(matches!(
decode_frame(&unknown_type),
Err(FirehoseError::UnknownType(value)) if value == "#unknown"
));
let relay_error = frame(
EventHeader { op: -1, t: None },
&serde_json::json!({
"error": "ConsumerTooSlow",
"message": "retry later",
}),
);
assert!(matches!(
decode_frame(&relay_error),
Err(FirehoseError::RelayError { error, message })
if error == "ConsumerTooSlow" && message.as_deref() == Some("retry later")
));
}
#[test]
fn firehose_frame_preserves_outdated_cursor_info() {
let bytes = frame(
EventHeader {
op: 1,
t: Some("#info".to_owned()),
},
&Info {
message: Some("cursor is outside the retained window".into()),
name: InfoName::OutdatedCursor,
},
);
let SubscribeReposMessage::Info(info) = decode_frame(&bytes).unwrap() else {
panic!("expected info");
};
assert_eq!(info.name, InfoName::OutdatedCursor);
}
#[test]
fn firehose_frame_maps_future_cursor_error() {
let bytes = frame(
EventHeader { op: -1, t: None },
&serde_json::json!({
"error": "FutureCursor",
"message": "requested cursor is ahead of the relay",
}),
);
assert!(matches!(
decode_frame(&bytes),
Err(FirehoseError::FutureCursor { message })
if message.as_deref() == Some("requested cursor is ahead of the relay")
));
}
}