syncbat 0.9.0

use batpak::event::HashChain;
use batpak::event::StoredEvent;
use batpak::id::EntityIdType;
use batpak::store::IndexEntry;
use serde::{Deserialize, Serialize};

use super::error::SubscriptionRuntimeError;

/// Canonical event-stream payload envelope encoded with `batpak::canonical::to_bytes`.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct EventStreamEnvelopeV1 {
    /// Envelope schema version.
    pub schema_version: u32,
    /// Globally unique subscription id.
    pub subscription_id: String,
    /// Committed event id.
    pub event_id: u128,
    /// Correlation id from the event header.
    pub correlation_id: u128,
    /// Optional causation id from the event header.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub causation_id: Option<u128>,
    /// Entity coordinate string.
    pub entity: String,
    /// Scope coordinate string.
    pub scope: String,
    /// Raw event kind u16.
    pub event_kind_raw: u16,
    /// Exported 4-bit event category.
    pub event_category: u8,
    /// Payload schema version stamped on the event header.
    pub payload_version: u16,
    /// Event header timestamp in microseconds.
    pub timestamp_us: i64,
    /// HLC wall milliseconds from the index entry.
    pub hlc_wall_ms: u64,
    /// Commit-order sequence for the event.
    pub global_sequence: u64,
    /// Payload content hash from the event header.
    pub content_hash: [u8; 32],
    /// Previous hash from the entity hash chain, if present.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub prev_hash: Option<[u8; 32]>,
    /// Event hash from the entity hash chain, if present.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub event_hash: Option<[u8; 32]>,
    /// Optional inner payload schema ref declared by the route.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub inner_event_payload_schema_ref: Option<String>,
    /// Raw committed payload bytes.
    pub payload: Vec<u8>,
}

impl EventStreamEnvelopeV1 {
    /// Build and canonically encode an envelope for one committed event.
    ///
    /// Reads the payload through the key-aware delivery primitive
    /// (`read_delivery_stored`): a readable event yields `Ok(Some(bytes))`
    /// carrying its PLAINTEXT payload; a crypto-shredded event yields `Ok(None)`
    /// so the caller skips it and never ships ciphertext into the public
    /// envelope. Without `payload-encryption` this is byte-identical to a raw read.
    ///
    /// # Errors
    /// [`SubscriptionRuntimeError::Store`] or [`SubscriptionRuntimeError::EnvelopeEncoding`].
    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())
        );
    }
}

/// Canonical projection-stream payload envelope encoded with `batpak::canonical::to_bytes`.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ProjectionStreamEnvelopeV1 {
    /// Envelope schema version.
    pub schema_version: u32,
    /// Globally unique subscription id.
    pub subscription_id: String,
    /// Route-declared projection id.
    pub projection_id: String,
    /// Entity coordinate watched by the projection.
    pub entity: String,
    /// Entity generation at which the state was materialized.
    pub entity_generation: u64,
    /// Freshness mode used to materialize the projection.
    pub freshness: String,
    /// Optional inner projection schema ref declared by the route.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub inner_projection_schema_ref: Option<String>,
    /// Blake3 digest of the canonical projection state bytes.
    pub state_hash: [u8; 32],
    /// Canonical bytes of the typed projection state.
    pub state: Vec<u8>,
}

impl ProjectionStreamEnvelopeV1 {
    /// Build and canonically encode an envelope for one projection update.
    ///
    /// # Errors
    /// [`SubscriptionRuntimeError::EnvelopeEncoding`].
    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(),
        )),
    }
}

/// Canonical operation-status-stream payload envelope encoded with `batpak::canonical::to_bytes`.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct OperationStatusStreamEnvelopeV1 {
    /// Envelope schema version.
    pub schema_version: u32,
    /// Globally unique subscription id.
    pub subscription_id: String,
    /// Route-declared operation name.
    pub operation: String,
    /// Entity coordinate watched for operation-status facts.
    pub entity: String,
    /// Entity generation at which the status was materialized.
    pub entity_generation: u64,
    /// Freshness mode used to materialize the status view.
    pub freshness: String,
    /// Optional inner status schema ref declared by the route.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub inner_status_schema_ref: Option<String>,
    /// Blake3 digest of the canonical status bytes.
    pub status_hash: [u8; 32],
    /// Canonical bytes of the typed operation-status view.
    pub status: Vec<u8>,
}

impl OperationStatusStreamEnvelopeV1 {
    /// Build and canonically encode an envelope for one operation-status update.
    ///
    /// # Errors
    /// [`SubscriptionRuntimeError::EnvelopeEncoding`].
    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()))
    }
}

/// Canonical receipt-stream payload envelope encoded with `batpak::canonical::to_bytes`.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ReceiptStreamEnvelopeV1 {
    /// Envelope schema version.
    pub schema_version: u32,
    /// Globally unique subscription id.
    pub subscription_id: String,
    /// Route-declared receipt kind filter.
    pub receipt_kind: String,
    /// Stable operation descriptor name from the receipt envelope.
    pub descriptor_name: String,
    /// Stable outcome class from the receipt envelope.
    pub outcome_class: String,
    /// Committed event id.
    pub event_id: u128,
    /// Correlation id from the event header.
    pub correlation_id: u128,
    /// Optional causation id from the event header.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub causation_id: Option<u128>,
    /// Entity coordinate string.
    pub entity: String,
    /// Scope coordinate string.
    pub scope: String,
    /// Payload schema version stamped on the event header.
    pub payload_version: u16,
    /// Event header timestamp in microseconds.
    pub timestamp_us: i64,
    /// HLC wall milliseconds from the index entry.
    pub hlc_wall_ms: u64,
    /// Commit-order sequence for the receipt event.
    pub global_sequence: u64,
    /// Payload content hash from the event header.
    pub content_hash: [u8; 32],
    /// Previous hash from the entity hash chain, if present.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub prev_hash: Option<[u8; 32]>,
    /// Event hash from the entity hash chain, if present.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub event_hash: Option<[u8; 32]>,
    /// Optional inner receipt schema ref declared by the route.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub inner_receipt_schema_ref: Option<String>,
    /// Blake3 digest of the canonical receipt envelope bytes.
    pub receipt_hash: [u8; 32],
    /// Canonical bytes of the typed [`crate::ReceiptEnvelope`].
    pub receipt: Vec<u8>,
}

impl ReceiptStreamEnvelopeV1 {
    /// Build and canonically encode an envelope for one committed receipt event.
    ///
    /// Reads the payload through the key-aware delivery primitive
    /// (`read_delivery_stored`): `Ok(None)` means the event was crypto-shredded
    /// and the caller must skip it, never shipping ciphertext. Without
    /// `payload-encryption` this is byte-identical to a raw read.
    ///
    /// # Errors
    /// [`SubscriptionRuntimeError::Store`], [`SubscriptionRuntimeError::EnvelopeEncoding`],
    /// or receipt payload decode failures surfaced as envelope encoding errors.
    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)
    }
}

/// Canonical entity-stream payload envelope encoded with `batpak::canonical::to_bytes`.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct EntityStreamEnvelopeV1 {
    /// Envelope schema version.
    pub schema_version: u32,
    /// Globally unique subscription id.
    pub subscription_id: String,
    /// Committed event id.
    pub event_id: u128,
    /// Correlation id from the event header.
    pub correlation_id: u128,
    /// Optional causation id from the event header.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub causation_id: Option<u128>,
    /// Entity coordinate string.
    pub entity: String,
    /// Scope coordinate string.
    pub scope: String,
    /// Raw event kind u16.
    pub event_kind_raw: u16,
    /// Exported 4-bit event category.
    pub event_category: u8,
    /// Payload schema version stamped on the event header.
    pub payload_version: u16,
    /// Event header timestamp in microseconds.
    pub timestamp_us: i64,
    /// HLC wall milliseconds from the index entry.
    pub hlc_wall_ms: u64,
    /// Commit-order sequence for the event.
    pub global_sequence: u64,
    /// Payload content hash from the event header.
    pub content_hash: [u8; 32],
    /// Previous hash from the entity hash chain, if present.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub prev_hash: Option<[u8; 32]>,
    /// Event hash from the entity hash chain, if present.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub event_hash: Option<[u8; 32]>,
    /// Optional inner payload schema ref declared by the route.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub inner_event_payload_schema_ref: Option<String>,
    /// Raw committed payload bytes.
    pub payload: Vec<u8>,
}

impl EntityStreamEnvelopeV1 {
    /// Build and canonically encode an envelope for one committed entity event.
    ///
    /// Reads the payload through the key-aware delivery primitive
    /// (`read_delivery_stored`): a readable event yields `Ok(Some(bytes))`
    /// carrying its PLAINTEXT payload; a crypto-shredded event yields `Ok(None)`
    /// so the caller skips it and never ships ciphertext into the public
    /// envelope. Without `payload-encryption` this is byte-identical to a raw read.
    ///
    /// # Errors
    /// [`SubscriptionRuntimeError::Store`] or [`SubscriptionRuntimeError::EnvelopeEncoding`].
    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(),
        }
    }
}

// ─── Key-aware delivery (crypto-shred, opt-in `payload-encryption`) ────────────
//
// The three raw-event stream sessions (event, entity, receipt) build a delivered
// envelope from a STORED event's payload. Under Stage C an encrypted event's
// on-disk payload is ciphertext, so shipping the stored bytes would ship
// ciphertext downstream — undecryptable, silent data loss. These helpers read the
// payload through the core-boundary decrypt primitive
// ([`batpak::store::Store::read_delivery_payload`], keys never crossing to
// syncbat): a readable event yields its PLAINTEXT bytes; a crypto-shredded event
// yields `None`, so the session skips it LOUDLY and advances its cursor (delivery
// stays coherent — no stall, no ciphertext). In a build without the feature — or a
// store with no keyset — this is exactly `read_raw`, byte-identical.

/// Key-aware delivery read: `Some(stored)` carries the plaintext payload bytes;
/// `None` means the event's payload key was destroyed (crypto-shredded) and the
/// caller must skip it, never ship the ciphertext.
///
/// # Errors
/// [`SubscriptionRuntimeError::Store`] on an I/O / decode / authentication error.
#[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),
    }
}

/// Non-encryption build: delivery reads the stored bytes directly — plaintext,
/// byte-identical to the pre-crypto-shred path. Never reports a shredded skip.
///
/// # Errors
/// [`SubscriptionRuntimeError::Store`] on an I/O / decode error.
#[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)?))
}

/// Emit the observable warn for a crypto-shredded event skipped during
/// subscription delivery. The event is not delivered and the session's cursor
/// advances past it; the skip is LOUD (this warn), never silent.
///
/// Always present (the sessions reference it on their shredded-skip path in both
/// builds; without `payload-encryption` that path is simply never taken).
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)"
    );
}

/// Build the canonical event-stream envelope bytes from an already-read (and, in
/// an encryption build, already-decrypted) stored event.
///
/// # Errors
/// [`SubscriptionRuntimeError::EnvelopeEncoding`] on canonical-encode failure.
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()))
}

/// Build the canonical entity-stream envelope bytes from an already-read (and, in
/// an encryption build, already-decrypted) stored event.
///
/// # Errors
/// [`SubscriptionRuntimeError::EnvelopeEncoding`] on canonical-encode failure.
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()))
}

/// Build a receipt-stream envelope (and its canonical bytes) from an already-read
/// (and, in an encryption build, already-decrypted) stored event plus its decoded
/// [`crate::ReceiptEnvelope`]. Shared by [`ReceiptStreamEnvelopeV1::encode_for_entry`]
/// and the key-aware receipt-delivery path so both produce identical bytes.
///
/// # Errors
/// [`SubscriptionRuntimeError::EnvelopeEncoding`] on canonical-encode failure.
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))
}