liminal-server 0.22.0

//! Broker control requests on the connection's admitted Lys session.

use liminal::protocol::{CONVERSATION_REPLY_REQUESTED_FLAG, Frame, MessageEnvelope};

use super::service_route::{SERVICE_REFUSED_CODE, parse_request};
use super::state::{ConnectionProcessState, FrameAction};
use super::supervisor::ConnectionRuntime;
use super::worker_route::Request;
use crate::services::ServiceRequest;
use crate::services::broker::BrokerRefusal;

/// Returns the broker's JSON answer or named refusal on the request's conversation.
pub(super) fn broker_frame_response(
    runtime: &ConnectionRuntime,
    state: &ConnectionProcessState,
    request: Request,
    subject: &str,
    envelope: &MessageEnvelope,
) -> FrameAction {
    let Request {
        flags,
        stream_id,
        conversation_id,
    } = request;
    let refused = |message: String| {
        FrameAction::Respond(Frame::ConversationError {
            flags: 0,
            stream_id,
            conversation_id,
            reason_code: SERVICE_REFUSED_CODE,
            message: Some(message),
        })
    };
    let Some(bindings) = runtime.service_bindings() else {
        return refused("broker_not_configured".to_owned());
    };
    let (operation, payload) = match parse_request(&envelope.payload) {
        Ok(parsed) => parsed,
        Err(message) => return refused(message),
    };
    let answer = match bindings.serve_broker(
        state.lys_session.as_deref(),
        subject,
        &ServiceRequest {
            operation: &operation,
            payload: &payload,
        },
        super::apply::unix_seconds_now(),
    ) {
        Ok(answer) => answer,
        Err(BrokerRefusal::Decision(error)) => {
            return refused(match serde_json::to_string(&error.refusal) {
                Ok(message) => message,
                Err(encoding) => format!("refusal_encoding_failed: {encoding}; {}", error.cause),
            });
        }
        Err(error) => return refused(error.to_string()),
    };
    if flags & CONVERSATION_REPLY_REQUESTED_FLAG == 0 {
        return FrameAction::NoResponse;
    }
    FrameAction::Respond(Frame::ConversationMessage {
        flags: CONVERSATION_REPLY_REQUESTED_FLAG,
        stream_id,
        conversation_id,
        envelope: MessageEnvelope::new(
            envelope.schema_id,
            envelope.causal_context.clone(),
            answer.to_string().into_bytes(),
        ),
    })
}

#[cfg(feature = "test-support")]
impl super::supervisor::ConnectionSupervisor {
    /// Starts a socket fixture over the same broker bindings it observes.
    ///
    /// # Errors
    /// The supervisor could not start or install its authority.
    pub fn with_broker_for_tests(
        services: std::sync::Arc<super::services::LiminalConnectionServices>,
        admitter: std::sync::Arc<dyn crate::auth_lys::session::LysAdmitter>,
        network_id: String,
        bindings: std::sync::Arc<crate::services::production::ServiceBindings>,
    ) -> Result<Self, crate::ServerError> {
        Self::with_lys_and_fatal_shutdown(
            services,
            None,
            None,
            Some(super::supervisor::LysInstall {
                admitter,
                network_id,
                link_host: None,
                services: Some(bindings),
                membership: None,
            }),
            crate::config::LimitsConfig::default(),
            crate::server::shutdown::ShutdownHandle::new(),
        )
    }
}