use batpak::event::HashChain;
use batpak::event::StoredEvent;
use batpak::id::EntityIdType;
use batpak::store::IndexEntry;
use serde::{Deserialize, Serialize};
use super::error::SubscriptionRuntimeError;
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct EventStreamEnvelopeV1 {
pub schema_version: u32,
pub subscription_id: String,
pub event_id: u128,
pub correlation_id: u128,
#[serde(skip_serializing_if = "Option::is_none")]
pub causation_id: Option<u128>,
pub entity: String,
pub scope: String,
pub event_kind_raw: u16,
pub event_category: u8,
pub payload_version: u16,
pub timestamp_us: i64,
pub hlc_wall_ms: u64,
pub global_sequence: u64,
pub content_hash: [u8; 32],
#[serde(skip_serializing_if = "Option::is_none")]
pub prev_hash: Option<[u8; 32]>,
#[serde(skip_serializing_if = "Option::is_none")]
pub event_hash: Option<[u8; 32]>,
#[serde(skip_serializing_if = "Option::is_none")]
pub inner_event_payload_schema_ref: Option<String>,
pub payload: Vec<u8>,
}
impl EventStreamEnvelopeV1 {
pub fn encode_for_entry(
store: &batpak::store::Store<batpak::store::Open>,
subscription_id: &str,
entry: &IndexEntry,
inner_event_payload_schema_ref: Option<&str>,
) -> Result<Option<Vec<u8>>, SubscriptionRuntimeError> {
let Some(stored) = read_delivery_stored(store, entry.event_id())? else {
return Ok(None);
};
event_stream_envelope_bytes_from_stored(
subscription_id,
entry,
&stored,
inner_event_payload_schema_ref,
)
.map(Some)
}
fn from_stored(
subscription_id: &str,
entry: &IndexEntry,
stored: &batpak::event::StoredEvent<Vec<u8>>,
inner_event_payload_schema_ref: Option<&str>,
) -> Self {
let (prev_hash, event_hash) = hash_chain_parts(stored.event.hash_chain.as_ref());
Self {
schema_version: 1,
subscription_id: subscription_id.to_owned(),
event_id: entry.event_id().as_u128(),
correlation_id: entry.correlation_id(),
causation_id: entry.causation_id(),
entity: stored.coordinate.entity().to_owned(),
scope: stored.coordinate.scope().to_owned(),
event_kind_raw: entry.event_kind().as_raw_u16(),
event_category: entry.event_kind().category(),
payload_version: stored.event.header.payload_version,
timestamp_us: stored.event.header.timestamp_us,
hlc_wall_ms: entry.wall_ms(),
global_sequence: entry.global_sequence(),
content_hash: stored.event.header.content_hash,
prev_hash,
event_hash,
inner_event_payload_schema_ref: inner_event_payload_schema_ref.map(str::to_owned),
payload: stored.event.payload.clone(),
}
}
}
fn hash_chain_parts(chain: Option<&HashChain>) -> (Option<[u8; 32]>, Option<[u8; 32]>) {
chain
.map(|chain| (Some(chain.prev_hash), Some(chain.event_hash)))
.unwrap_or((None, None))
}
#[cfg(test)]
mod envelope_helper_tests {
use super::{blake3_state_hash, freshness_label, hash_chain_parts};
use batpak::event::HashChain;
use batpak::store::Freshness;
#[test]
fn hash_chain_parts_some_returns_exact_pair() {
let chain = HashChain {
prev_hash: [7_u8; 32],
event_hash: [9_u8; 32],
};
assert_eq!(
hash_chain_parts(Some(&chain)),
(Some([7_u8; 32]), Some([9_u8; 32]))
);
}
#[test]
fn hash_chain_parts_none_returns_none_pair() {
assert_eq!(hash_chain_parts(None), (None, None));
}
#[test]
fn blake3_state_hash_is_real_digest_not_constant() {
let hash = blake3_state_hash(b"projection-state");
assert_eq!(hash, *blake3::hash(b"projection-state").as_bytes());
assert_ne!(
blake3_state_hash(b"projection-state"),
blake3_state_hash(b"other-state")
);
assert_ne!(hash, [0_u8; 32]);
assert_ne!(hash, [1_u8; 32]);
}
#[test]
fn freshness_label_maps_each_supported_mode() {
assert_eq!(
freshness_label(&Freshness::Consistent).ok(),
Some("consistent".to_owned())
);
assert_eq!(
freshness_label(&Freshness::MaybeStale { max_stale_ms: 5 }).ok(),
Some("maybe_stale".to_owned())
);
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ProjectionStreamEnvelopeV1 {
pub schema_version: u32,
pub subscription_id: String,
pub projection_id: String,
pub entity: String,
pub entity_generation: u64,
pub freshness: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub inner_projection_schema_ref: Option<String>,
pub state_hash: [u8; 32],
pub state: Vec<u8>,
}
impl ProjectionStreamEnvelopeV1 {
pub fn encode<T>(
subscription_id: &str,
projection_id: &str,
entity: &str,
entity_generation: u64,
freshness: &batpak::store::Freshness,
inner_projection_schema_ref: Option<&str>,
state: &T,
) -> Result<Vec<u8>, SubscriptionRuntimeError>
where
T: serde::Serialize,
{
let state_bytes = batpak::canonical::to_bytes(state)
.map_err(|error| SubscriptionRuntimeError::EnvelopeEncoding(error.to_string()))?;
let state_hash = blake3_state_hash(&state_bytes);
let envelope = Self {
schema_version: 1,
subscription_id: subscription_id.to_owned(),
projection_id: projection_id.to_owned(),
entity: entity.to_owned(),
entity_generation,
freshness: freshness_label(freshness)?,
inner_projection_schema_ref: inner_projection_schema_ref.map(str::to_owned),
state_hash,
state: state_bytes,
};
batpak::canonical::to_bytes(&envelope)
.map_err(|error| SubscriptionRuntimeError::EnvelopeEncoding(error.to_string()))
}
}
fn blake3_state_hash(state_bytes: &[u8]) -> [u8; 32] {
*blake3::hash(state_bytes).as_bytes()
}
fn freshness_label(
freshness: &batpak::store::Freshness,
) -> Result<String, SubscriptionRuntimeError> {
match freshness {
batpak::store::Freshness::Consistent => Ok("consistent".to_owned()),
batpak::store::Freshness::MaybeStale { .. } => Ok("maybe_stale".to_owned()),
_ => Err(SubscriptionRuntimeError::EnvelopeEncoding(
"unsupported projection freshness mode".to_owned(),
)),
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct OperationStatusStreamEnvelopeV1 {
pub schema_version: u32,
pub subscription_id: String,
pub operation: String,
pub entity: String,
pub entity_generation: u64,
pub freshness: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub inner_status_schema_ref: Option<String>,
pub status_hash: [u8; 32],
pub status: Vec<u8>,
}
impl OperationStatusStreamEnvelopeV1 {
pub fn encode(
subscription_id: &str,
operation: &str,
entity: &str,
entity_generation: u64,
freshness: &batpak::store::Freshness,
inner_status_schema_ref: Option<&str>,
status: &crate::operation_status::OperationStatusView,
) -> Result<Vec<u8>, SubscriptionRuntimeError> {
let status_bytes = batpak::canonical::to_bytes(status)
.map_err(|error| SubscriptionRuntimeError::EnvelopeEncoding(error.to_string()))?;
let status_hash = blake3_state_hash(&status_bytes);
let envelope = Self {
schema_version: 1,
subscription_id: subscription_id.to_owned(),
operation: operation.to_owned(),
entity: entity.to_owned(),
entity_generation,
freshness: freshness_label(freshness)?,
inner_status_schema_ref: inner_status_schema_ref.map(str::to_owned),
status_hash,
status: status_bytes,
};
batpak::canonical::to_bytes(&envelope)
.map_err(|error| SubscriptionRuntimeError::EnvelopeEncoding(error.to_string()))
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ReceiptStreamEnvelopeV1 {
pub schema_version: u32,
pub subscription_id: String,
pub receipt_kind: String,
pub descriptor_name: String,
pub outcome_class: String,
pub event_id: u128,
pub correlation_id: u128,
#[serde(skip_serializing_if = "Option::is_none")]
pub causation_id: Option<u128>,
pub entity: String,
pub scope: String,
pub payload_version: u16,
pub timestamp_us: i64,
pub hlc_wall_ms: u64,
pub global_sequence: u64,
pub content_hash: [u8; 32],
#[serde(skip_serializing_if = "Option::is_none")]
pub prev_hash: Option<[u8; 32]>,
#[serde(skip_serializing_if = "Option::is_none")]
pub event_hash: Option<[u8; 32]>,
#[serde(skip_serializing_if = "Option::is_none")]
pub inner_receipt_schema_ref: Option<String>,
pub receipt_hash: [u8; 32],
pub receipt: Vec<u8>,
}
impl ReceiptStreamEnvelopeV1 {
pub fn encode_for_entry(
store: &batpak::store::Store<batpak::store::Open>,
subscription_id: &str,
route_receipt_kind: &str,
entry: &IndexEntry,
inner_receipt_schema_ref: Option<&str>,
) -> Result<Option<(Self, Vec<u8>)>, SubscriptionRuntimeError> {
let Some(stored) = read_delivery_stored(store, entry.event_id())? else {
return Ok(None);
};
let receipt: crate::ReceiptEnvelope = batpak::canonical::from_bytes(&stored.event.payload)
.map_err(|error| {
SubscriptionRuntimeError::EnvelopeEncoding(format!(
"receipt payload decode failed: {error}"
))
})?;
receipt_envelope_from_stored(
subscription_id,
route_receipt_kind,
entry,
&stored,
&receipt,
inner_receipt_schema_ref,
)
.map(Some)
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct EntityStreamEnvelopeV1 {
pub schema_version: u32,
pub subscription_id: String,
pub event_id: u128,
pub correlation_id: u128,
#[serde(skip_serializing_if = "Option::is_none")]
pub causation_id: Option<u128>,
pub entity: String,
pub scope: String,
pub event_kind_raw: u16,
pub event_category: u8,
pub payload_version: u16,
pub timestamp_us: i64,
pub hlc_wall_ms: u64,
pub global_sequence: u64,
pub content_hash: [u8; 32],
#[serde(skip_serializing_if = "Option::is_none")]
pub prev_hash: Option<[u8; 32]>,
#[serde(skip_serializing_if = "Option::is_none")]
pub event_hash: Option<[u8; 32]>,
#[serde(skip_serializing_if = "Option::is_none")]
pub inner_event_payload_schema_ref: Option<String>,
pub payload: Vec<u8>,
}
impl EntityStreamEnvelopeV1 {
pub fn encode_for_entry(
store: &batpak::store::Store<batpak::store::Open>,
subscription_id: &str,
entry: &IndexEntry,
inner_event_payload_schema_ref: Option<&str>,
) -> Result<Option<Vec<u8>>, SubscriptionRuntimeError> {
let Some(stored) = read_delivery_stored(store, entry.event_id())? else {
return Ok(None);
};
entity_stream_envelope_bytes_from_stored(
subscription_id,
entry,
&stored,
inner_event_payload_schema_ref,
)
.map(Some)
}
fn from_stored(
subscription_id: &str,
entry: &IndexEntry,
stored: &batpak::event::StoredEvent<Vec<u8>>,
inner_event_payload_schema_ref: Option<&str>,
) -> Self {
let (prev_hash, event_hash) = hash_chain_parts(stored.event.hash_chain.as_ref());
Self {
schema_version: 1,
subscription_id: subscription_id.to_owned(),
event_id: entry.event_id().as_u128(),
correlation_id: entry.correlation_id(),
causation_id: entry.causation_id(),
entity: stored.coordinate.entity().to_owned(),
scope: stored.coordinate.scope().to_owned(),
event_kind_raw: entry.event_kind().as_raw_u16(),
event_category: entry.event_kind().category(),
payload_version: stored.event.header.payload_version,
timestamp_us: stored.event.header.timestamp_us,
hlc_wall_ms: entry.wall_ms(),
global_sequence: entry.global_sequence(),
content_hash: stored.event.header.content_hash,
prev_hash,
event_hash,
inner_event_payload_schema_ref: inner_event_payload_schema_ref.map(str::to_owned),
payload: stored.event.payload.clone(),
}
}
}
#[cfg(feature = "payload-encryption")]
pub(super) fn read_delivery_stored(
store: &batpak::store::Store<batpak::store::Open>,
event_id: batpak::id::EventId,
) -> Result<Option<StoredEvent<Vec<u8>>>, SubscriptionRuntimeError> {
match store.read_delivery_payload(event_id)? {
batpak::store::DeliveryPayload::Readable(stored) => Ok(Some(*stored)),
batpak::store::DeliveryPayload::Shredded { .. } => Ok(None),
}
}
#[cfg(not(feature = "payload-encryption"))]
pub(super) fn read_delivery_stored(
store: &batpak::store::Store<batpak::store::Open>,
event_id: batpak::id::EventId,
) -> Result<Option<StoredEvent<Vec<u8>>>, SubscriptionRuntimeError> {
Ok(Some(store.read_raw(event_id)?))
}
pub(super) fn warn_shredded_delivery(
stream: &str,
subscription_id: &str,
event_id: batpak::id::EventId,
) {
tracing::warn!(
target: "syncbat::delivery",
stream,
subscription_id,
event_id = event_id.as_u128(),
"skipping a crypto-shredded event during subscription delivery; it is not delivered \
and the cursor advances past it (payload key destroyed — plaintext gone)"
);
}
pub(super) fn event_stream_envelope_bytes_from_stored(
subscription_id: &str,
entry: &IndexEntry,
stored: &StoredEvent<Vec<u8>>,
inner_event_payload_schema_ref: Option<&str>,
) -> Result<Vec<u8>, SubscriptionRuntimeError> {
let envelope = EventStreamEnvelopeV1::from_stored(
subscription_id,
entry,
stored,
inner_event_payload_schema_ref,
);
batpak::canonical::to_bytes(&envelope)
.map_err(|error| SubscriptionRuntimeError::EnvelopeEncoding(error.to_string()))
}
pub(super) fn entity_stream_envelope_bytes_from_stored(
subscription_id: &str,
entry: &IndexEntry,
stored: &StoredEvent<Vec<u8>>,
inner_event_payload_schema_ref: Option<&str>,
) -> Result<Vec<u8>, SubscriptionRuntimeError> {
let envelope = EntityStreamEnvelopeV1::from_stored(
subscription_id,
entry,
stored,
inner_event_payload_schema_ref,
);
batpak::canonical::to_bytes(&envelope)
.map_err(|error| SubscriptionRuntimeError::EnvelopeEncoding(error.to_string()))
}
pub(super) fn receipt_envelope_from_stored(
subscription_id: &str,
route_receipt_kind: &str,
entry: &IndexEntry,
stored: &StoredEvent<Vec<u8>>,
receipt: &crate::ReceiptEnvelope,
inner_receipt_schema_ref: Option<&str>,
) -> Result<(ReceiptStreamEnvelopeV1, Vec<u8>), SubscriptionRuntimeError> {
let receipt_bytes = batpak::canonical::to_bytes(receipt)
.map_err(|error| SubscriptionRuntimeError::EnvelopeEncoding(error.to_string()))?;
let receipt_hash = blake3_state_hash(&receipt_bytes);
let (prev_hash, event_hash) = hash_chain_parts(stored.event.hash_chain.as_ref());
let envelope = ReceiptStreamEnvelopeV1 {
schema_version: 1,
subscription_id: subscription_id.to_owned(),
receipt_kind: route_receipt_kind.to_owned(),
descriptor_name: receipt.descriptor_name.clone(),
outcome_class: receipt.outcome.class().to_owned(),
event_id: entry.event_id().as_u128(),
correlation_id: entry.correlation_id(),
causation_id: entry.causation_id(),
entity: stored.coordinate.entity().to_owned(),
scope: stored.coordinate.scope().to_owned(),
payload_version: stored.event.header.payload_version,
timestamp_us: stored.event.header.timestamp_us,
hlc_wall_ms: entry.wall_ms(),
global_sequence: entry.global_sequence(),
content_hash: stored.event.header.content_hash,
prev_hash,
event_hash,
inner_receipt_schema_ref: inner_receipt_schema_ref.map(str::to_owned),
receipt_hash,
receipt: receipt_bytes,
};
let envelope_bytes = batpak::canonical::to_bytes(&envelope)
.map_err(|error| SubscriptionRuntimeError::EnvelopeEncoding(error.to_string()))?;
Ok((envelope, envelope_bytes))
}