traverse_runtime/events/

broker.rs

//! Synchronous in-process event broker.
//!
//! Governed by spec 026-event-broker and spec 036-event-subscription-replay.

use std::{
    collections::{HashMap, HashSet, VecDeque},
    sync::{Arc, Mutex},
    time::Duration,
};

use super::{
    catalog::EventCatalog,
    types::{
        BrokerEvent, EventBroker, EventCursor, EventError, LifecycleStatus, Subscription,
        SubscriptionId, SubscriptionPoll, TraverseEvent,
    },
    validation::{EventValidationEvidence, EventValidationMode, validate_event},
};

/// Clock abstraction used by the broker for retention pruning.
pub trait BrokerClock: Send + Sync {
    fn now(&self) -> std::time::SystemTime;
}

#[derive(Debug)]
pub struct SystemClock;

impl BrokerClock for SystemClock {
    fn now(&self) -> std::time::SystemTime {
        std::time::SystemTime::now()
    }
}

/// Broker runtime configuration.
#[derive(Debug, Clone)]
pub struct BrokerConfig {
    pub retention_window: Duration,
    pub max_queue_len: usize,
}

impl Default for BrokerConfig {
    fn default() -> Self {
        Self {
            retention_window: Duration::from_mins(5),
            max_queue_len: 1024,
        }
    }
}

#[derive(Debug, Clone)]
struct BufferedEvent {
    cursor: u64,
    published_at: std::time::SystemTime,
    event: TraverseEvent,
}

#[derive(Debug)]
struct SubscriptionState {
    subscription_id: SubscriptionId,
    event_type: String,
    subject_id: Option<String>,
    consumer_id: Option<String>,
    cursor: u64,
    queue: VecDeque<BufferedEvent>,
}

#[derive(Debug, Default)]
struct BrokerState {
    next_subscription: u64,
    next_cursor: HashMap<String, u64>,
    buffers: HashMap<String, VecDeque<BufferedEvent>>,
    seen_event_ids: HashMap<String, HashSet<String>>,
    subscriptions: HashMap<SubscriptionId, SubscriptionState>,
    subscriptions_by_event_type: HashMap<String, HashSet<SubscriptionId>>,
    /// See [`EventBroker::seed_restart_floor`].
    restart_floor: u64,
    validation_evidence: Vec<EventValidationEvidence>,
    quarantine_records: Vec<EventQuarantineRecord>,
    observed_lineage: Vec<EventLineageRecord>,
    telemetry: Vec<EventTelemetryRecord>,
    metrics: EventRuntimeMetrics,
}

/// Sanitized observation that a broker delivered one governed event.
///
/// This is runtime evidence, not a catalog declaration. It intentionally
/// excludes the event payload and authenticated subject data.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EventLineageRecord {
    pub contract_id: String,
    pub contract_version: String,
    pub event_id: String,
    pub producer_id: String,
    pub consumer_id: String,
    pub subscription_id: SubscriptionId,
    pub cursor: EventCursor,
}

/// Sanitized record written when enforcement rejects an event envelope.
///
/// The record is deliberately distinct from migration diagnostics: it models
/// the governed quarantine stream without retaining the rejected payload.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EventQuarantineRecord {
    pub evidence: EventValidationEvidence,
}

/// Portable, OpenTelemetry-compatible event boundary evidence.
///
/// Host adapters can map this stable, payload-free shape to their chosen
/// OpenTelemetry SDK. The in-process broker retains it for deterministic
/// conformance tests and never exports an event payload or subject identity.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EventTelemetryRecord {
    pub operation: &'static str,
    pub outcome: &'static str,
    pub contract_id: String,
    pub contract_version: String,
    pub event_id: String,
    pub deduplication_id: Option<String>,
    pub ordering_scope: Option<String>,
    pub correlation_id: Option<String>,
    pub causation_id: Option<String>,
    pub consumer_id: Option<String>,
    pub cursor: Option<EventCursor>,
    pub retry_count: u32,
    pub latency_ms: u64,
}

/// Counter snapshot for OpenTelemetry-compatible event runtime evidence.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct EventRuntimeMetrics {
    pub publications: u64,
    pub deliveries: u64,
    pub validation_failures: u64,
    pub quarantines: u64,
}

/// Synchronous, in-memory implementation of [`EventBroker`].
///
/// The broker stores a bounded retention buffer per event type and maintains a
/// bounded delivery queue per subscription. Subscribers poll for events using a
/// broker-issued subscription id and a cursor.
pub struct InProcessBroker {
    catalog: Arc<EventCatalog>,
    config: BrokerConfig,
    clock: Arc<dyn BrokerClock>,
    state: Mutex<BrokerState>,
    validation_mode: EventValidationMode,
}

impl std::fmt::Debug for InProcessBroker {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("InProcessBroker").finish_non_exhaustive()
    }
}

impl InProcessBroker {
    /// Create a new broker backed by the given catalog.
    ///
    /// # Errors
    ///
    /// Returns [`EventError::InvalidRetentionWindow`] when the provided configuration is invalid.
    pub fn new(catalog: Arc<EventCatalog>) -> Result<Self, EventError> {
        Self::with_clock(catalog, BrokerConfig::default(), Arc::new(SystemClock))
    }

    /// Create a broker with explicit configuration and clock.
    ///
    /// # Errors
    ///
    /// Returns [`EventError::InvalidRetentionWindow`] when the provided configuration is invalid.
    pub fn with_clock(
        catalog: Arc<EventCatalog>,
        config: BrokerConfig,
        clock: Arc<dyn BrokerClock>,
    ) -> Result<Self, EventError> {
        Self::with_clock_and_validation(catalog, config, clock, EventValidationMode::Migration)
    }

    /// Create a broker with an explicit governed-event enforcement policy.
    ///
    /// Migration mode records violations without interrupting existing traffic;
    /// enforcement mode rejects invalid envelopes.
    ///
    /// # Errors
    ///
    /// Returns [`EventError::InvalidRetentionWindow`] when the configuration
    /// cannot maintain replay and queue guarantees.
    pub fn with_clock_and_validation(
        catalog: Arc<EventCatalog>,
        config: BrokerConfig,
        clock: Arc<dyn BrokerClock>,
        validation_mode: EventValidationMode,
    ) -> Result<Self, EventError> {
        if config.retention_window == Duration::from_secs(0) {
            return Err(EventError::InvalidRetentionWindow(
                "retention_window must be > 0".to_string(),
            ));
        }
        if config.max_queue_len == 0 {
            return Err(EventError::InvalidRetentionWindow(
                "max_queue_len must be > 0".to_string(),
            ));
        }

        Ok(Self {
            catalog,
            config,
            clock,
            state: Mutex::new(BrokerState::default()),
            validation_mode,
        })
    }

    /// Returns sanitized validation and quarantine evidence. Payload data is
    /// never retained through this interface.
    #[must_use]
    pub fn validation_evidence(&self) -> Vec<EventValidationEvidence> {
        self.state
            .lock()
            .map(|state| state.validation_evidence.clone())
            .unwrap_or_default()
    }

    /// Returns sanitized enforcement rejections prepared for governed quarantine.
    #[must_use]
    pub fn quarantine_records(&self) -> Vec<EventQuarantineRecord> {
        self.state
            .lock()
            .map(|state| state.quarantine_records.clone())
            .unwrap_or_default()
    }

    /// Returns sanitized runtime delivery observations for catalog reconciliation.
    #[must_use]
    pub fn observed_lineage(&self) -> Vec<EventLineageRecord> {
        self.state
            .lock()
            .map(|state| state.observed_lineage.clone())
            .unwrap_or_default()
    }

    /// Returns deterministic, sanitized boundary telemetry for host export.
    #[must_use]
    pub fn telemetry(&self) -> Vec<EventTelemetryRecord> {
        self.state
            .lock()
            .map(|state| state.telemetry.clone())
            .unwrap_or_default()
    }

    /// Returns a deterministic counter snapshot for event runtime evidence.
    #[must_use]
    pub fn metrics(&self) -> EventRuntimeMetrics {
        self.state
            .lock()
            .map(|state| state.metrics.clone())
            .unwrap_or_default()
    }

    fn subscribe_with_subject(
        &self,
        event_type: &str,
        from_cursor: &str,
        subject_id: Option<&str>,
        consumer_id: Option<&str>,
    ) -> Result<Subscription, EventError> {
        if self.catalog.get(event_type).is_none() {
            return Err(EventError::UnregisteredEventType(event_type.to_owned()));
        }

        let from_cursor = parse_cursor(from_cursor)?;
        let now = self.clock.now();
        let mut state = self
            .state
            .lock()
            .map_err(|_| EventError::LifecycleViolation("broker lock poisoned".to_owned()))?;
        prune_expired(&mut state, event_type, self.config.retention_window, now);
        validate_from_cursor(&state, event_type, from_cursor)?;
        self.catalog.increment_consumer_count(event_type);

        state.next_subscription = state.next_subscription.saturating_add(1);
        let subscription_id = format!("sub-{}", state.next_subscription);
        let mut queue = VecDeque::new();
        for item in state
            .buffers
            .get(event_type)
            .into_iter()
            .flat_map(|buffer| buffer.iter())
        {
            if (from_cursor == 0 || item.cursor > from_cursor)
                && subject_id
                    .is_none_or(|subject| item.event.subject_id.as_deref() == Some(subject))
            {
                enqueue_with_drop_oldest(&mut queue, self.config.max_queue_len, item.clone());
            }
        }

        state.subscriptions.insert(
            subscription_id.clone(),
            SubscriptionState {
                subscription_id: subscription_id.clone(),
                event_type: event_type.to_string(),
                subject_id: subject_id.map(str::to_owned),
                consumer_id: consumer_id.map(str::to_owned),
                cursor: from_cursor,
                queue,
            },
        );
        state
            .subscriptions_by_event_type
            .entry(event_type.to_string())
            .or_default()
            .insert(subscription_id.clone());

        Ok(Subscription {
            subscription_id,
            event_type: event_type.to_string(),
            cursor: cursor_to_string(from_cursor),
        })
    }

    /// Subscribe with the consuming capability identity needed for observed lineage.
    ///
    /// The identity is runtime evidence only; it does not declare a catalog relationship.
    ///
    /// # Errors
    ///
    /// Returns [`EventError::LifecycleViolation`] when `consumer_id` is empty,
    /// plus the same errors as [`EventBroker::subscribe_for_subject`].
    pub fn subscribe_for_consumer(
        &self,
        event_type: &str,
        from_cursor: &str,
        consumer_id: &str,
        subject_id: Option<&str>,
    ) -> Result<Subscription, EventError> {
        if consumer_id.trim().is_empty() {
            return Err(EventError::LifecycleViolation(
                "consumer_id must not be empty".to_string(),
            ));
        }
        self.subscribe_with_subject(event_type, from_cursor, subject_id, Some(consumer_id))
    }

    fn validate_boundary(&self, event: &TraverseEvent) -> Result<(), EventError> {
        let validation = validate_event(event, self.validation_mode);
        let validation_outcome = if validation.is_valid() {
            "accepted"
        } else if validation.accepted {
            "reported"
        } else {
            "rejected"
        };
        let evidence = EventValidationEvidence::from_result(&validation);
        let mut state = self
            .state
            .lock()
            .map_err(|_| EventError::LifecycleViolation("broker lock poisoned".to_owned()))?;
        if let Some(evidence) = evidence {
            state.validation_evidence.push(evidence.clone());
            state.metrics.validation_failures = state.metrics.validation_failures.saturating_add(1);
            if !validation.accepted {
                state
                    .quarantine_records
                    .push(EventQuarantineRecord { evidence });
                state.metrics.quarantines = state.metrics.quarantines.saturating_add(1);
            }
        }
        state.telemetry.push(telemetry_record(
            "traverse.event.validation",
            validation_outcome,
            event,
            None,
            None,
        ));
        if !validation.accepted {
            let code = validation
                .diagnostics
                .first()
                .map_or("EVP-000", |diagnostic| diagnostic.code);
            return Err(EventError::ValidationRejected(code.to_owned()));
        }
        Ok(())
    }

    /// Shared implementation for `publish` and `publish_with_cursor`.
    /// `assigned_cursor`, when given, is adopted as this event's cursor
    /// instead of self-incrementing the per-type counter (spec 066 FR-007:
    /// keeps live-delivery cursors consistent with the durable journal's
    /// cursor space). The per-type counter still tracks the highest cursor
    /// ever used so `validate_from_cursor`'s empty-buffer fallback remains
    /// correct regardless of cursor source.
    fn publish_internal(
        &self,
        event: &TraverseEvent,
        assigned_cursor: Option<u64>,
    ) -> Result<(), EventError> {
        self.validate_boundary(event)?;
        let entry = self
            .catalog
            .get(&event.event_type)
            .ok_or_else(|| EventError::UnregisteredEventType(event.event_type.clone()))?;

        match entry.lifecycle_status {
            LifecycleStatus::Active => {}
            LifecycleStatus::Deprecated => {
                return Err(EventError::LifecycleViolation(format!(
                    "event type '{}' is Deprecated and cannot be published",
                    event.event_type
                )));
            }
            LifecycleStatus::Draft => {
                return Err(EventError::LifecycleViolation(format!(
                    "event type '{}' is Draft and cannot be published",
                    event.event_type
                )));
            }
        }

        let now = self.clock.now();

        let mut state = self
            .state
            .lock()
            .map_err(|_| EventError::LifecycleViolation("broker lock poisoned".to_owned()))?;

        prune_expired(
            &mut state,
            &event.event_type,
            self.config.retention_window,
            now,
        );

        let seen = state
            .seen_event_ids
            .entry(event.event_type.clone())
            .or_default();
        if seen.contains(&event.id) {
            // Duplicate emissions are silently discarded.
            return Ok(());
        }
        seen.insert(event.id.clone());
        state.metrics.publications = state.metrics.publications.saturating_add(1);
        state.telemetry.push(telemetry_record(
            "traverse.event.publish",
            "accepted",
            event,
            None,
            None,
        ));

        let next = state
            .next_cursor
            .entry(event.event_type.clone())
            .or_insert(0);
        let cursor = if let Some(assigned) = assigned_cursor {
            *next = (*next).max(assigned);
            assigned
        } else {
            *next = next.saturating_add(1);
            *next
        };

        let buffered = BufferedEvent {
            cursor,
            published_at: now,
            event: event.clone(),
        };

        state
            .buffers
            .entry(event.event_type.clone())
            .or_default()
            .push_back(buffered.clone());

        let subscription_ids = subscription_ids_for_event_type(&state, &event.event_type);
        for subscription_id in subscription_ids {
            let Some(sub) = state.subscriptions.get_mut(&subscription_id) else {
                continue;
            };
            if sub
                .subject_id
                .as_deref()
                .is_some_and(|subject_id| event.subject_id.as_deref() != Some(subject_id))
            {
                continue;
            }
            enqueue_with_drop_oldest(&mut sub.queue, self.config.max_queue_len, buffered.clone());
        }

        Ok(())
    }
}

fn parse_cursor(raw: &str) -> Result<u64, EventError> {
    let trimmed = raw.trim();
    if trimmed == "0" {
        return Ok(0);
    }
    trimmed.parse::<u64>().map_err(|_| {
        EventError::InvalidCursor("cursor must be \"0\" or a base-10 unsigned integer".to_string())
    })
}

fn cursor_to_string(cursor: u64) -> EventCursor {
    cursor.to_string()
}

fn enqueue_with_drop_oldest(
    queue: &mut VecDeque<BufferedEvent>,
    max_len: usize,
    item: BufferedEvent,
) {
    while queue.len() >= max_len {
        let _ = queue.pop_front();
    }
    queue.push_back(item);
}

fn prune_expired(
    state: &mut BrokerState,
    event_type: &str,
    retention_window: Duration,
    now: std::time::SystemTime,
) {
    let buffer = state.buffers.entry(event_type.to_string()).or_default();
    let mut oldest_retained_cursor = None;
    while let Some(front) = buffer.pop_front() {
        let age = now
            .duration_since(front.published_at)
            .unwrap_or(Duration::from_secs(0));
        if age <= retention_window {
            oldest_retained_cursor = Some(front.cursor);
            buffer.push_front(front);
            break;
        }

        if let Some(ids) = state.seen_event_ids.get_mut(event_type) {
            let _ = ids.remove(&front.event.id);
        }
    }

    let Some(oldest_cursor) = oldest_retained_cursor else {
        // Buffer is empty after pruning; nothing to sync.
        return;
    };

    // Sync per-subscription queues so they don't deliver events that are no longer retained.
    let subscription_ids = subscription_ids_for_event_type(state, event_type);
    for subscription_id in subscription_ids {
        let Some(sub) = state.subscriptions.get_mut(&subscription_id) else {
            continue;
        };
        while let Some(front) = sub.queue.front() {
            if front.cursor >= oldest_cursor {
                break;
            }
            let _ = sub.queue.pop_front();
        }
        if sub.cursor != 0 && sub.cursor < oldest_cursor.saturating_sub(1) {
            // Cursor is now outside the retention window; keep it as-is so poll can surface cursor_expired.
        }
    }
}

fn validate_from_cursor(
    state: &BrokerState,
    event_type: &str,
    from_cursor: u64,
) -> Result<(), EventError> {
    if from_cursor == 0 {
        return Ok(());
    }

    let last_cursor = state
        .next_cursor
        .get(event_type)
        .copied()
        .unwrap_or(0)
        .max(state.restart_floor);
    if let Some(buffer) = state.buffers.get(event_type)
        && let Some(front) = buffer.front()
    {
        let oldest_ok = front.cursor.saturating_sub(1);
        if from_cursor < oldest_ok {
            return Err(EventError::CursorExpired {
                event_type: event_type.to_string(),
                oldest_available_cursor: cursor_to_string(oldest_ok),
            });
        }
        return Ok(());
    }

    // If the buffer is empty but we have published events before, treat cursors behind the last
    // observed cursor as expired to avoid silent gaps.
    if last_cursor > 0 && from_cursor < last_cursor {
        return Err(EventError::CursorExpired {
            event_type: event_type.to_string(),
            oldest_available_cursor: cursor_to_string(last_cursor),
        });
    }

    Ok(())
}

fn subscription_ids_for_event_type(
    state: &BrokerState,
    event_type: &str,
) -> HashSet<SubscriptionId> {
    state
        .subscriptions_by_event_type
        .get(event_type)
        .cloned()
        .unwrap_or_default()
}

impl EventBroker for InProcessBroker {
    fn subscribe_for_subject(
        &self,
        event_type: &str,
        from_cursor: &str,
        subject_id: Option<&str>,
    ) -> Result<Subscription, EventError> {
        self.subscribe_with_subject(event_type, from_cursor, subject_id, None)
    }

    fn seed_restart_floor(&self, floor: u64) {
        if let Ok(mut state) = self.state.lock() {
            state.restart_floor = state.restart_floor.max(floor);
        }
    }

    /// Publish `event` to all registered subscribers.
    ///
    /// # Errors
    ///
    /// - [`EventError::UnregisteredEventType`] if the event type is not in the catalog.
    /// - [`EventError::LifecycleViolation`] if the catalog entry is `Draft` or `Deprecated`.
    fn publish(&self, event: TraverseEvent) -> Result<(), EventError> {
        self.publish_internal(&event, None)
    }

    /// Publish `event`, adopting `cursor` as this broker's own cursor for it
    /// instead of self-assigning the next per-type sequence value. Used by
    /// [`DurableBroker`](super::durable::DurableBroker) so live-delivery
    /// cursors stay numerically consistent with the durable journal's cursor
    /// space (spec 066 FR-007): a cursor obtained during live polling
    /// remains valid when a later `poll` falls back to durable replay.
    ///
    /// # Errors
    ///
    /// Returns [`EventError::InvalidCursor`] when `cursor` is not a base-10
    /// unsigned integer, plus the same errors as [`Self::publish`].
    fn publish_with_cursor(&self, event: TraverseEvent, cursor: &str) -> Result<(), EventError> {
        let assigned = parse_cursor(cursor)?;
        self.publish_internal(&event, Some(assigned))
    }

    /// Create a subscription for `event_type` starting from `from_cursor`.
    ///
    /// The event type must already be registered in the catalog.
    ///
    /// # Errors
    ///
    /// Returns [`EventError::UnregisteredEventType`] if the event type is not catalogued.
    fn subscribe(&self, event_type: &str, from_cursor: &str) -> Result<Subscription, EventError> {
        self.subscribe_with_subject(event_type, from_cursor, None, None)
    }

    /// Poll a subscription for up to `max_events`.
    ///
    /// # Errors
    ///
    fn poll(
        &self,
        subscription_id: &str,
        max_events: usize,
    ) -> Result<SubscriptionPoll, EventError> {
        let now = self.clock.now();
        let mut state = self
            .state
            .lock()
            .map_err(|_| EventError::LifecycleViolation("broker lock poisoned".to_owned()))?;

        let mut subscription = state
            .subscriptions
            .remove(subscription_id)
            .ok_or_else(|| EventError::SubscriptionNotFound(subscription_id.to_string()))?;
        let event_type = subscription.event_type.clone();
        let cursor = subscription.cursor;

        prune_expired(&mut state, &event_type, self.config.retention_window, now);

        validate_from_cursor(&state, &event_type, cursor)?;

        if let Some(buffer) = state.buffers.get(&event_type)
            && let Some(oldest_cursor) = buffer.front().map(|e| e.cursor)
        {
            while let Some(front) = subscription.queue.front() {
                if front.cursor >= oldest_cursor {
                    break;
                }
                let _ = subscription.queue.pop_front();
            }
        }

        if max_events == 0 {
            let cursor_str = cursor_to_string(subscription.cursor);
            state
                .subscriptions
                .insert(subscription.subscription_id.clone(), subscription);
            return Ok(SubscriptionPoll {
                subscription_id: subscription_id.to_string(),
                event_type,
                cursor: cursor_str,
                events: Vec::new(),
            });
        }

        let mut out = Vec::new();
        let mut delivered_cursor = subscription.cursor;
        for _ in 0..max_events {
            let Some(item) = subscription.queue.pop_front() else {
                break;
            };
            delivered_cursor = item.cursor;
            state.observed_lineage.push(EventLineageRecord {
                contract_id: item.event.event_type.clone(),
                contract_version: item.event.version.clone(),
                event_id: item.event.id.clone(),
                producer_id: item.event.owner.clone(),
                consumer_id: subscription
                    .consumer_id
                    .clone()
                    .unwrap_or_else(|| subscription.subscription_id.clone()),
                subscription_id: subscription.subscription_id.clone(),
                cursor: cursor_to_string(item.cursor),
            });
            let consumer_id = subscription
                .consumer_id
                .clone()
                .unwrap_or_else(|| subscription.subscription_id.clone());
            state.telemetry.push(telemetry_record(
                "traverse.event.delivery",
                "delivered",
                &item.event,
                Some(consumer_id),
                Some(cursor_to_string(item.cursor)),
            ));
            out.push(BrokerEvent {
                cursor: cursor_to_string(item.cursor),
                event: item.event,
            });
        }
        subscription.cursor = delivered_cursor;
        state.metrics.deliveries = state.metrics.deliveries.saturating_add(out.len() as u64);

        let subscription_id_value = subscription.subscription_id.clone();
        let event_type_value = subscription.event_type.clone();
        let cursor_value = cursor_to_string(subscription.cursor);
        state
            .subscriptions
            .insert(subscription.subscription_id.clone(), subscription);

        Ok(SubscriptionPoll {
            subscription_id: subscription_id_value,
            event_type: event_type_value,
            cursor: cursor_value,
            events: out,
        })
    }

    /// Cancel a subscription.
    ///
    /// # Errors
    ///
    /// Returns [`EventError::SubscriptionNotFound`] if the subscription id is unknown.
    fn cancel(&self, subscription_id: &str) -> Result<(), EventError> {
        let mut state = self
            .state
            .lock()
            .map_err(|_| EventError::LifecycleViolation("broker lock poisoned".to_owned()))?;

        let Some(subscription) = state.subscriptions.remove(subscription_id) else {
            return Err(EventError::SubscriptionNotFound(
                subscription_id.to_string(),
            ));
        };
        if let Some(ids) = state
            .subscriptions_by_event_type
            .get_mut(&subscription.event_type)
        {
            let _ = ids.remove(subscription_id);
            if ids.is_empty() {
                let _ = state
                    .subscriptions_by_event_type
                    .remove(&subscription.event_type);
            }
        }
        Ok(())
    }
}

fn telemetry_record(
    operation: &'static str,
    outcome: &'static str,
    event: &TraverseEvent,
    consumer_id: Option<String>,
    cursor: Option<EventCursor>,
) -> EventTelemetryRecord {
    EventTelemetryRecord {
        operation,
        outcome,
        contract_id: event.event_type.clone(),
        contract_version: event.version.clone(),
        event_id: event.id.clone(),
        deduplication_id: event.deduplication_id.clone(),
        ordering_scope: event.ordering_scope.clone(),
        correlation_id: event.correlation_id.clone(),
        causation_id: event.causation_id.clone(),
        consumer_id,
        cursor,
        retry_count: 0,
        latency_ms: 0,
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::expect_used)]
    #![allow(clippy::panic)]
    #![allow(clippy::unwrap_used)]

    use super::*;
    use crate::events::catalog::EventCatalogEntry;

    fn cursor_expired_oldest(err: &EventError) -> Option<String> {
        if let EventError::CursorExpired {
            oldest_available_cursor,
            ..
        } = err
        {
            Some(oldest_available_cursor.clone())
        } else {
            None
        }
    }

    fn make_catalog(event_type: &str, status: LifecycleStatus) -> Arc<EventCatalog> {
        let catalog = Arc::new(EventCatalog::new());
        catalog
            .register(EventCatalogEntry {
                event_type: event_type.to_string(),
                owner: "cap.test".to_string(),
                version: "1.0.0".to_string(),
                lifecycle_status: status,
                consumer_count: 0,
            })
            .expect("catalog register must succeed");
        catalog
    }

    fn sample_event(event_type: &str, id: &str) -> TraverseEvent {
        TraverseEvent {
            id: id.to_string(),
            source: "traverse-runtime/cap.test".to_string(),
            event_type: event_type.to_string(),
            datacontenttype: "application/json".to_string(),
            time: "2026-04-08T00:00:00Z".to_string(),
            data: serde_json::json!({}),
            owner: "cap.test".to_string(),
            version: "1.0.0".to_string(),
            lifecycle_status: LifecycleStatus::Active,
            deduplication_id: Some(id.to_string()),
            ordering_scope: Some("test".to_string()),
            correlation_id: Some("correlation-test".to_string()),
            causation_id: Some("command-test".to_string()),
            subject_id: None,
            actor_id: None,
        }
    }

    #[test]
    fn broker_debug_impl_is_accessible() {
        let catalog = make_catalog("dev.traverse.debug", LifecycleStatus::Active);
        let broker = InProcessBroker::new(catalog).expect("broker must be created");
        let rendered = format!("{broker:?}");
        assert!(rendered.contains("InProcessBroker"));
    }

    #[test]
    fn invalid_max_queue_len_is_rejected() {
        let catalog = make_catalog("dev.traverse.invalid", LifecycleStatus::Active);
        let err = InProcessBroker::with_clock(
            catalog,
            BrokerConfig {
                retention_window: Duration::from_secs(1),
                max_queue_len: 0,
            },
            Arc::new(SystemClock),
        )
        .expect_err("max_queue_len=0 must be rejected");
        assert!(matches!(err, EventError::InvalidRetentionWindow(_)));
    }

    #[test]
    fn publish_with_cursor_adopts_the_given_cursor_instead_of_self_assigning() {
        let event_type = "dev.traverse.injected-cursor";
        let catalog = make_catalog(event_type, LifecycleStatus::Active);
        let broker = InProcessBroker::new(catalog).expect("broker must be created");

        broker
            .publish_with_cursor(sample_event(event_type, "evt-1"), "42")
            .expect("publish_with_cursor must succeed");

        let subscription = broker
            .subscribe(event_type, "0")
            .expect("subscribe must succeed");
        let poll = broker
            .poll(&subscription.subscription_id, 10)
            .expect("poll must succeed");
        assert_eq!(poll.events.len(), 1);
        assert_eq!(poll.events[0].cursor, "42");
        assert_eq!(poll.cursor, "42");
    }

    #[test]
    fn publish_with_cursor_rejects_a_malformed_cursor() {
        let event_type = "dev.traverse.injected-cursor-invalid";
        let catalog = make_catalog(event_type, LifecycleStatus::Active);
        let broker = InProcessBroker::new(catalog).expect("broker must be created");

        let err = broker
            .publish_with_cursor(sample_event(event_type, "evt-1"), "not-a-cursor")
            .expect_err("malformed cursor must be rejected");
        assert!(matches!(err, EventError::InvalidCursor(_)));
    }

    #[test]
    fn default_publish_with_cursor_ignores_the_cursor_and_self_assigns() {
        struct SelfAssigningOnlyBroker(InProcessBroker);

        impl EventBroker for SelfAssigningOnlyBroker {
            fn publish(&self, event: TraverseEvent) -> Result<(), EventError> {
                self.0.publish(event)
            }
            fn subscribe(
                &self,
                event_type: &str,
                from_cursor: &str,
            ) -> Result<Subscription, EventError> {
                self.0.subscribe(event_type, from_cursor)
            }
            fn subscribe_for_subject(
                &self,
                event_type: &str,
                from_cursor: &str,
                subject_id: Option<&str>,
            ) -> Result<Subscription, EventError> {
                self.0
                    .subscribe_for_subject(event_type, from_cursor, subject_id)
            }
            fn poll(
                &self,
                subscription_id: &str,
                max_events: usize,
            ) -> Result<SubscriptionPoll, EventError> {
                self.0.poll(subscription_id, max_events)
            }
            fn cancel(&self, subscription_id: &str) -> Result<(), EventError> {
                self.0.cancel(subscription_id)
            }
        }

        let event_type = "dev.traverse.default-publish-with-cursor";
        let catalog = make_catalog(event_type, LifecycleStatus::Active);
        let broker =
            SelfAssigningOnlyBroker(InProcessBroker::new(catalog).expect("broker must be created"));

        // The default `publish_with_cursor` ignores the supplied cursor and
        // behaves exactly like `publish` (self-assigning cursor "1").
        broker
            .publish_with_cursor(sample_event(event_type, "evt-1"), "999")
            .expect("default publish_with_cursor must succeed");

        let subscription = broker
            .subscribe(event_type, "0")
            .expect("subscribe must succeed");
        let poll = broker
            .poll(&subscription.subscription_id, 10)
            .expect("poll must succeed");
        assert_eq!(poll.events[0].cursor, "1");

        // The default `seed_restart_floor` is a no-op; exercise it alongside
        // this wrapper's other pass-through trait methods.
        broker.seed_restart_floor(999);
        let subject_subscription = broker
            .subscribe_for_subject(event_type, "0", None)
            .expect("subscribe_for_subject must succeed");
        broker
            .cancel(&subject_subscription.subscription_id)
            .expect("cancel must succeed");
    }

    #[test]
    fn invalid_cursor_is_rejected() {
        let catalog = make_catalog("dev.traverse.cursor", LifecycleStatus::Active);
        let broker = InProcessBroker::new(catalog).expect("broker must be created");
        let err = broker
            .subscribe("dev.traverse.cursor", "not-a-cursor")
            .expect_err("invalid cursor must fail");
        assert!(matches!(err, EventError::InvalidCursor(_)));
    }

    #[test]
    fn subject_subscription_filters_backlog_and_live_delivery() {
        let event_type = "dev.traverse.subject-filter";
        let broker = InProcessBroker::new(make_catalog(event_type, LifecycleStatus::Active))
            .expect("broker must be created");
        let mut other = sample_event(event_type, "evt-other");
        other.subject_id = Some("subject-other".to_string());
        let mut expected = sample_event(event_type, "evt-match");
        expected.subject_id = Some("subject-match".to_string());
        broker.publish(other).expect("backlog publish must succeed");
        broker
            .publish(expected.clone())
            .expect("backlog publish must succeed");

        let subscription = broker
            .subscribe_for_subject(event_type, "0", Some("subject-match"))
            .expect("subject subscription must succeed");
        let backlog = broker
            .poll(&subscription.subscription_id, 10)
            .expect("backlog poll must succeed");
        assert_eq!(backlog.events.len(), 1);
        assert_eq!(backlog.events[0].event.id, expected.id);

        let mut live_other = sample_event(event_type, "evt-live-other");
        live_other.subject_id = Some("subject-other".to_string());
        let mut live_expected = sample_event(event_type, "evt-live-match");
        live_expected.subject_id = Some("subject-match".to_string());
        broker
            .publish(live_other)
            .expect("non-matching live publish must succeed");
        broker
            .publish(live_expected.clone())
            .expect("matching live publish must succeed");

        let live = broker
            .poll(&subscription.subscription_id, 10)
            .expect("live poll must succeed");
        assert_eq!(live.events.len(), 1);
        assert_eq!(live.events[0].event.id, live_expected.id);
    }

    #[test]
    fn consumer_subscription_records_sanitized_observed_lineage() {
        let event_type = "dev.traverse.lineage.observed";
        let broker = InProcessBroker::new(make_catalog(event_type, LifecycleStatus::Active))
            .expect("broker must be created");
        let subscription = broker
            .subscribe_for_consumer(event_type, "0", "capability.audit", None)
            .expect("consumer subscription must succeed");
        let mut event = sample_event(event_type, "evt-lineage");
        event.owner = "capability.orders".to_string();
        event.version = "1.2.3".to_string();
        event.data = serde_json::json!({"secret":"not lineage"});
        broker.publish(event).expect("publish must succeed");

        let _ = broker
            .poll(&subscription.subscription_id, 1)
            .expect("poll must succeed");
        assert_eq!(
            broker.observed_lineage(),
            vec![EventLineageRecord {
                contract_id: event_type.to_string(),
                contract_version: "1.2.3".to_string(),
                event_id: "evt-lineage".to_string(),
                producer_id: "capability.orders".to_string(),
                consumer_id: "capability.audit".to_string(),
                subscription_id: subscription.subscription_id,
                cursor: "1".to_string(),
            }]
        );
        assert_eq!(
            broker.metrics(),
            EventRuntimeMetrics {
                publications: 1,
                deliveries: 1,
                validation_failures: 0,
                quarantines: 0,
            }
        );
        let telemetry = broker.telemetry();
        assert_eq!(telemetry.len(), 3);
        assert_eq!(telemetry[0].operation, "traverse.event.validation");
        assert_eq!(telemetry[0].outcome, "accepted");
        assert_eq!(telemetry[1].operation, "traverse.event.publish");
        assert_eq!(telemetry[2].operation, "traverse.event.delivery");
        assert_eq!(
            telemetry[2].consumer_id.as_deref(),
            Some("capability.audit")
        );
        assert_eq!(telemetry[2].cursor.as_deref(), Some("1"));
        assert_eq!(telemetry[2].contract_version, "1.2.3");
        assert_eq!(telemetry[2].retry_count, 0);
        assert_eq!(telemetry[2].latency_ms, 0);
        assert!(!format!("{telemetry:?}").contains("not lineage"));
    }

    #[test]
    fn consumer_subscription_rejects_empty_identity() {
        let event_type = "dev.traverse.lineage.identity";
        let broker = InProcessBroker::new(make_catalog(event_type, LifecycleStatus::Active))
            .expect("broker must be created");
        let err = broker
            .subscribe_for_consumer(event_type, "0", " ", None)
            .expect_err("empty consumer identity must fail");
        assert!(matches!(err, EventError::LifecycleViolation(_)));
    }

    #[test]
    fn quarantine_records_fail_closed_when_broker_state_is_poisoned() {
        let broker = InProcessBroker::new(make_catalog(
            "dev.traverse.quarantine.poison",
            LifecycleStatus::Active,
        ))
        .expect("broker must be created");
        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _guard = broker.state.lock().expect("state lock must be available");
            panic!("poison state lock");
        }));

        assert!(broker.quarantine_records().is_empty());
        assert_eq!(broker.metrics(), EventRuntimeMetrics::default());
    }

    #[test]
    fn publish_rejects_deprecated_and_draft_event_types() {
        let deprecated = InProcessBroker::new(make_catalog(
            "dev.traverse.deprecated",
            LifecycleStatus::Deprecated,
        ))
        .expect("broker must be created");
        let err = deprecated
            .publish(sample_event("dev.traverse.deprecated", "evt-001"))
            .expect_err("deprecated publish must fail");
        assert!(matches!(err, EventError::LifecycleViolation(_)));

        let draft =
            InProcessBroker::new(make_catalog("dev.traverse.draft", LifecycleStatus::Draft))
                .expect("broker must be created");
        let err = draft
            .publish(sample_event("dev.traverse.draft", "evt-001"))
            .expect_err("draft publish must fail");
        assert!(matches!(err, EventError::LifecycleViolation(_)));
    }

    #[test]
    fn enforcement_rejects_invalid_events_and_retains_sanitized_evidence() {
        let event_type = "dev.traverse.orders.created";
        let broker = InProcessBroker::with_clock_and_validation(
            make_catalog(event_type, LifecycleStatus::Active),
            BrokerConfig::default(),
            Arc::new(SystemClock),
            EventValidationMode::Enforcement,
        )
        .expect("broker must be created");
        let mut invalid = sample_event(event_type, "evt-invalid");
        invalid.owner.clear();
        invalid.data = serde_json::json!({"customer_email": "private@example.test"});

        let error = broker
            .publish(invalid)
            .expect_err("enforcement must reject a missing owner");
        assert!(matches!(error, EventError::ValidationRejected(code) if code == "EVP-005"));
        let evidence = broker.validation_evidence();
        assert_eq!(evidence.len(), 1);
        assert_eq!(evidence[0].contract_id, event_type);
        assert_eq!(evidence[0].diagnostics[0].code, "EVP-005");
        assert!(!format!("{evidence:?}").contains("private@example.test"));
        let quarantine = broker.quarantine_records();
        assert_eq!(quarantine.len(), 1);
        assert_eq!(quarantine[0].evidence, evidence[0]);
        assert!(!format!("{quarantine:?}").contains("private@example.test"));
        assert_eq!(
            broker.metrics(),
            EventRuntimeMetrics {
                publications: 0,
                deliveries: 0,
                validation_failures: 1,
                quarantines: 1,
            }
        );
    }

    #[test]
    fn migration_records_invalid_event_evidence_without_rejecting_delivery() {
        let event_type = "dev.traverse.orders.created";
        let broker = InProcessBroker::new(make_catalog(event_type, LifecycleStatus::Active))
            .expect("broker must be created");
        let mut invalid = sample_event(event_type, "evt-migration");
        invalid.owner.clear();

        broker
            .publish(invalid)
            .expect("migration mode must preserve delivery");
        assert_eq!(broker.validation_evidence().len(), 1);
        assert!(broker.quarantine_records().is_empty());
        assert_eq!(broker.metrics().validation_failures, 1);
        assert_eq!(broker.metrics().quarantines, 0);
    }

    #[test]
    fn broker_lock_poisoning_surfaces_lifecycle_violation() {
        let broker =
            InProcessBroker::new(make_catalog("dev.traverse.poison", LifecycleStatus::Active))
                .expect("broker must be created");

        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _guard = broker.state.lock().unwrap();
            panic!("poison lock");
        }));

        let err = broker
            .publish(sample_event("dev.traverse.poison", "evt-001"))
            .expect_err("poisoned publish must fail");
        assert!(matches!(err, EventError::LifecycleViolation(_)));

        let err = broker
            .subscribe("dev.traverse.poison", "0")
            .expect_err("poisoned subscribe must fail");
        assert!(matches!(err, EventError::LifecycleViolation(_)));

        let err = broker
            .poll("sub-1", 1)
            .expect_err("poisoned poll must fail");
        assert!(matches!(err, EventError::LifecycleViolation(_)));

        let err = broker
            .cancel("sub-1")
            .expect_err("poisoned cancel must fail");
        assert!(matches!(err, EventError::LifecycleViolation(_)));
    }

    #[derive(Debug)]
    struct ManualClock(std::sync::Mutex<std::time::SystemTime>);

    impl ManualClock {
        fn new(now: std::time::SystemTime) -> Self {
            Self(std::sync::Mutex::new(now))
        }

        fn advance(&self, by: Duration) {
            if let Ok(mut guard) = self.0.lock()
                && let Some(next) = guard.checked_add(by)
            {
                *guard = next;
            }
        }

        fn set(&self, now: std::time::SystemTime) {
            if let Ok(mut guard) = self.0.lock() {
                *guard = now;
            }
        }
    }

    impl BrokerClock for ManualClock {
        fn now(&self) -> std::time::SystemTime {
            self.0
                .lock()
                .ok()
                .map_or(std::time::SystemTime::UNIX_EPOCH, |guard| *guard)
        }
    }

    #[test]
    fn clock_regression_does_not_break_retention_pruning() {
        let clock = Arc::new(ManualClock::new(std::time::SystemTime::UNIX_EPOCH));
        let broker = InProcessBroker::with_clock(
            make_catalog("dev.traverse.clock", LifecycleStatus::Active),
            BrokerConfig {
                retention_window: Duration::from_mins(1),
                max_queue_len: 16,
            },
            clock.clone(),
        )
        .expect("broker must be created");

        clock.set(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(10));
        broker
            .publish(sample_event("dev.traverse.clock", "evt-001"))
            .expect("publish must succeed");

        // Move time backwards to force duration_since() to hit the error path.
        clock.set(std::time::SystemTime::UNIX_EPOCH);
        broker
            .publish(sample_event("dev.traverse.clock", "evt-002"))
            .expect("publish must succeed");
    }

    #[test]
    fn publish_pruning_syncs_subscription_queues_and_skips_other_event_types() {
        let catalog = Arc::new(EventCatalog::new());
        catalog
            .register(EventCatalogEntry {
                event_type: "dev.traverse.a".to_string(),
                owner: "cap.test".to_string(),
                version: "1.0.0".to_string(),
                lifecycle_status: LifecycleStatus::Active,
                consumer_count: 0,
            })
            .expect("register must succeed");
        catalog
            .register(EventCatalogEntry {
                event_type: "dev.traverse.b".to_string(),
                owner: "cap.test".to_string(),
                version: "1.0.0".to_string(),
                lifecycle_status: LifecycleStatus::Active,
                consumer_count: 0,
            })
            .expect("register must succeed");

        let clock = Arc::new(ManualClock::new(std::time::SystemTime::UNIX_EPOCH));
        let broker = InProcessBroker::with_clock(
            catalog,
            BrokerConfig {
                retention_window: Duration::from_secs(5),
                max_queue_len: 64,
            },
            clock.clone(),
        )
        .expect("broker must be created");

        let sub_a = broker
            .subscribe("dev.traverse.a", "1")
            .expect("subscribe must succeed");
        let sub_b = broker
            .subscribe("dev.traverse.b", "0")
            .expect("subscribe must succeed");

        broker
            .publish(sample_event("dev.traverse.a", "evt-001"))
            .expect("publish must succeed");
        clock.advance(Duration::from_secs(1));
        broker
            .publish(sample_event("dev.traverse.a", "evt-002"))
            .expect("publish must succeed");
        clock.advance(Duration::from_secs(1));
        broker
            .publish(sample_event("dev.traverse.a", "evt-003"))
            .expect("publish must succeed");

        // Jump forward so evt-001 and evt-002 are outside retention; evt-003 is retained.
        clock.advance(Duration::from_secs(5));
        broker
            .publish(sample_event("dev.traverse.a", "evt-004"))
            .expect("publish must succeed");

        let err = broker
            .poll(&sub_a.subscription_id, 10)
            .expect_err("poll must surface cursor_expired after retention pruning");
        let oldest_available_cursor = cursor_expired_oldest(&err).expect("must be cursor_expired");

        let sub_a_resumed = broker
            .subscribe("dev.traverse.a", &oldest_available_cursor)
            .expect("subscribe must succeed");
        let poll_a = broker
            .poll(&sub_a_resumed.subscription_id, 10)
            .expect("poll must succeed");
        assert!(
            poll_a
                .events
                .first()
                .is_some_and(|e| e.event.id == "evt-003"),
            "queue must resume from oldest retained event"
        );

        let poll_b = broker
            .poll(&sub_b.subscription_id, 10)
            .expect("poll must succeed");
        assert!(
            poll_b.events.is_empty(),
            "event_type mismatch must not enqueue"
        );

        // Also cover the non-cursor_expired branch in the extraction logic above.
        let other_err = broker
            .poll("sub-missing", 10)
            .expect_err("poll must fail when subscription is missing");
        assert!(cursor_expired_oldest(&other_err).is_none());
    }

    #[test]
    fn subscribe_replays_events_from_existing_buffer() {
        let clock = Arc::new(ManualClock::new(std::time::SystemTime::UNIX_EPOCH));
        let broker = InProcessBroker::with_clock(
            make_catalog("dev.traverse.replay", LifecycleStatus::Active),
            BrokerConfig {
                retention_window: Duration::from_secs(5),
                max_queue_len: 64,
            },
            clock,
        )
        .expect("broker must be created");

        broker
            .publish(sample_event("dev.traverse.replay", "evt-001"))
            .expect("publish must succeed");

        let sub = broker
            .subscribe("dev.traverse.replay", "0")
            .expect("subscribe must succeed");
        let poll = broker
            .poll(&sub.subscription_id, 10)
            .expect("poll must succeed");
        assert_eq!(poll.events.len(), 1);
        assert_eq!(poll.events[0].event.id, "evt-001");
    }

    #[test]
    fn subscribe_rejects_cursor_expired_when_buffer_non_empty() {
        let clock = Arc::new(ManualClock::new(std::time::SystemTime::UNIX_EPOCH));
        let broker = InProcessBroker::with_clock(
            make_catalog("dev.traverse.expire", LifecycleStatus::Active),
            BrokerConfig {
                retention_window: Duration::from_secs(5),
                max_queue_len: 64,
            },
            clock.clone(),
        )
        .expect("broker must be created");

        for i in 1..=5 {
            broker
                .publish(sample_event("dev.traverse.expire", &format!("evt-{i:03}")))
                .expect("publish must succeed");
            clock.advance(Duration::from_secs(1));
        }

        // Advance so only the last event remains within retention.
        clock.advance(Duration::from_secs(5));

        let err = broker
            .subscribe("dev.traverse.expire", "1")
            .expect_err("subscribe must fail with cursor_expired");
        assert!(matches!(err, EventError::CursorExpired { .. }));
    }

    #[test]
    fn poll_with_zero_max_events_returns_empty() {
        let broker =
            InProcessBroker::new(make_catalog("dev.traverse.poll0", LifecycleStatus::Active))
                .expect("broker must be created");
        let sub = broker
            .subscribe("dev.traverse.poll0", "0")
            .expect("subscribe must succeed");
        let poll = broker
            .poll(&sub.subscription_id, 0)
            .expect("poll must succeed");
        assert!(poll.events.is_empty());
    }

    #[test]
    fn poll_prunes_subscription_queue_based_on_retention() {
        let clock = Arc::new(ManualClock::new(std::time::SystemTime::UNIX_EPOCH));
        let broker = InProcessBroker::with_clock(
            make_catalog("dev.traverse.pollprune", LifecycleStatus::Active),
            BrokerConfig {
                retention_window: Duration::from_secs(5),
                max_queue_len: 64,
            },
            clock.clone(),
        )
        .expect("broker must be created");

        let sub = broker
            .subscribe("dev.traverse.pollprune", "0")
            .expect("subscribe must succeed");

        broker
            .publish(sample_event("dev.traverse.pollprune", "evt-001"))
            .expect("publish must succeed");
        clock.advance(Duration::from_secs(4));
        broker
            .publish(sample_event("dev.traverse.pollprune", "evt-002"))
            .expect("publish must succeed");

        // Advance so evt-001 is outside retention but evt-002 is retained.
        clock.advance(Duration::from_secs(3));

        let poll = broker
            .poll(&sub.subscription_id, 10)
            .expect("poll must succeed");
        assert_eq!(poll.events.len(), 1);
        assert_eq!(poll.events[0].event.id, "evt-002");
    }

    #[test]
    fn cancel_unknown_subscription_returns_not_found() {
        let broker = InProcessBroker::new(make_catalog(
            "dev.traverse.cancel-miss",
            LifecycleStatus::Active,
        ))
        .expect("broker must be created");
        let err = broker.cancel("sub-missing").expect_err("cancel must fail");
        assert!(matches!(err, EventError::SubscriptionNotFound(_)));
    }

    #[test]
    fn publish_tolerates_a_stale_event_type_index_entry() {
        let event_type = "dev.traverse.stale-index";
        let broker = InProcessBroker::new(make_catalog(event_type, LifecycleStatus::Active))
            .expect("broker must be created");
        let subscription = broker
            .subscribe(event_type, "0")
            .expect("subscribe must succeed");

        broker
            .state
            .lock()
            .expect("broker lock must be available")
            .subscriptions
            .remove(&subscription.subscription_id);

        broker
            .publish(sample_event(event_type, "evt-stale-index"))
            .expect("stale index entry must not prevent publication");
    }

    #[test]
    fn cancel_removes_an_empty_event_type_index() {
        let event_type = "dev.traverse.cancel-index";
        let broker = InProcessBroker::new(make_catalog(event_type, LifecycleStatus::Active))
            .expect("broker must be created");
        let subscription = broker
            .subscribe(event_type, "0")
            .expect("subscribe must succeed");

        broker
            .cancel(&subscription.subscription_id)
            .expect("cancel must succeed");

        let state = broker.state.lock().expect("broker lock must be available");
        assert!(!state.subscriptions_by_event_type.contains_key(event_type));
    }

    #[test]
    fn cancel_tolerates_a_missing_event_type_index_entry() {
        let event_type = "dev.traverse.cancel-missing-index";
        let broker = InProcessBroker::new(make_catalog(event_type, LifecycleStatus::Active))
            .expect("broker must be created");
        let subscription = broker
            .subscribe(event_type, "0")
            .expect("subscribe must succeed");

        broker
            .state
            .lock()
            .expect("broker lock must be available")
            .subscriptions_by_event_type
            .remove(event_type);

        broker
            .cancel(&subscription.subscription_id)
            .expect("missing index entry must not prevent cancellation");
    }
}