liminal-server 0.23.1

use super::{ConnectionServices, LimitsConfig, ServerError};

/// ACCESS-003 R1: the context watch drains its own subscription the way the
/// connection pump does; each reading carries its message id, the stable
/// event id Lys charges it under.
impl crate::services::context_watch::ContextFeed for super::services::ConnectionSubscription {
    fn next_reading(&mut self) -> Option<(String, Vec<u8>)> {
        self.try_next()
            .map(|envelope| (envelope.message_id.as_uuid().to_string(), envelope.payload))
    }
}

/// Starts the context watch over the `context-watch` channel when the
/// service bindings ask for it (ACCESS-003 R1).
///
/// One subscription whose inbox
/// wakes the watcher on every admitted reading, bounded by the same §5
/// limits a connection's subscription is. `Ok(None)` when the bindings do not
/// watch.
///
/// # Errors
/// The `context-watch` channel refused the subscription (it must be
/// configured), or the watcher thread did not start.
pub fn start_context_watch(
    services: &dyn ConnectionServices,
    bindings: &crate::services::production::ServiceBindings,
    limits: &LimitsConfig,
) -> Result<Option<crate::services::context_watch::ContextWatch>, ServerError> {
    if !bindings.watches_context() {
        return Ok(None);
    }
    let wake = crate::services::context_watch::ContextWake::new();
    let install = liminal::channel::InboxInstall {
        budget: liminal::channel::ConnectionInboxBudget::new(limits.max_connection_inbox_bytes),
        depth_cap: limits.max_subscription_inbox_depth,
        notifier: Some(wake.notifier()),
        capacity: None,
    };
    let subscription = services.subscribe(
        crate::services::ServiceKind::ContextWatch.subject_shape(),
        &[],
        Some(install),
    )?;
    crate::services::context_watch::ContextWatch::start(
        subscription,
        wake,
        bindings.context_reporter(),
    )
    .map(Some)
    .map_err(|error| ServerError::ConfigValidation {
        message: error.to_string(),
    })
}