liminal-server 0.25.4

//! The service request route (ACCESS-003 R1): a `ConversationMessage` on a
//! conversation open on a service subject (`seats.<name>`, `budgets.<name>`,
//! `alarms`, `launches`, `context-watch`, `sessions.<id>.keys`) is a request
//! to that service, served on the connection's admitted Lys session. A hot
//! service is decided offline for `call` on the subject; a deliberate one
//! (`sessions.<id>.keys`) asks Lys live and is held under a draft or acted
//! once. The outcome rides back on the same conversation when the request
//! asked for a reply; a refusal is the conversation's error frame, by name.
//!
//! The request payload is a JSON object naming the caller's operation id,
//! the idempotency key every effect is made under, with the service's own
//! body beside it:
//!
//! ```json
//! {"operation": "op-7f3a", "payload": {"...": "..."}}
//! ```
//!
//! `payload` is handed to the service as its JSON bytes, empty when absent.
//! The reply carries the request's schema id and a JSON body: `acted` with
//! the grant, `held` with the mode, grant and draft, or `policy` with Lys's
//! budget snapshot.

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

use super::state::{ConnectionProcessState, FrameAction};
use super::supervisor::runtime::ConnectionRuntime;
use crate::services::{ServiceOutcome, ServiceRefusal, ServiceRequest, serve_session};

/// The reason code a refused service request carries on its
/// `ConversationError` (ACCESS-003 R1); the message names the refusal.
pub(super) const SERVICE_REFUSED_CODE: u16 = 0x0301;

/// Serves the request `envelope` carries on the service `subject` the
/// conversation was opened on, deciding it on the connection's admitted
/// session, and answers the outcome or the refusal on the conversation.
pub(super) fn service_frame_response(
    runtime: &ConnectionRuntime,
    state: &ConnectionProcessState,
    flags: u8,
    stream_id: u32,
    conversation_id: u64,
    subject: &str,
    envelope: &MessageEnvelope,
) -> FrameAction {
    let refused = |message: String| {
        FrameAction::Respond(Frame::ConversationError {
            flags: 0,
            stream_id,
            conversation_id,
            reason_code: SERVICE_REFUSED_CODE,
            message: Some(message),
        })
    };
    let Some((_, network_id)) = runtime.lys_admitter() else {
        return refused("no lys admitter is installed; a service subject needs one".to_owned());
    };
    let Some(bindings) = runtime.service_bindings() else {
        return refused(
            "no service bindings are installed ([auth.lys.services]); the service cannot act"
                .to_owned(),
        );
    };
    let (operation, payload) = match parse_request(&envelope.payload) {
        Ok(parsed) => parsed,
        Err(message) => return refused(message),
    };
    let request = ServiceRequest {
        operation: &operation,
        payload: &payload,
    };
    let outcome = serve_session(
        &bindings.services(),
        state.lys_session.as_deref(),
        network_id,
        subject,
        &request,
        super::apply::unix_seconds_now(),
    );
    match outcome {
        Ok(outcome) => {
            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(),
                    outcome_body(&outcome),
                ),
            })
        }
        Err(ServiceRefusal::Decision(error)) => refused(
            serde_json::to_string(&error.refusal).unwrap_or_else(|_| error.cause.to_string()),
        ),
        Err(refusal) => refused(refusal.to_string()),
    }
}

/// The operation id and the service body of a request payload; refused by
/// name when the payload is not a JSON object naming an operation.
pub(super) fn parse_request(payload: &[u8]) -> Result<(String, Vec<u8>), String> {
    let value: serde_json::Value = serde_json::from_slice(payload)
        .map_err(|error| format!("service request is not JSON: {error}"))?;
    let Some(object) = value.as_object() else {
        return Err("service request is not a JSON object naming an operation".to_owned());
    };
    let operation = object
        .get("operation")
        .and_then(serde_json::Value::as_str)
        .map(str::trim)
        .filter(|operation| !operation.is_empty())
        .ok_or_else(|| "service request names no operation id; nothing was done".to_owned())?
        .to_owned();
    let body = match object.get("payload") {
        Some(body) => serde_json::to_vec(body)
            .map_err(|error| format!("service request payload could not be carried: {error}"))?,
        None => Vec::new(),
    };
    Ok((operation, body))
}

/// The JSON reply body of an outcome.
fn outcome_body(outcome: &ServiceOutcome) -> Vec<u8> {
    let body = match outcome {
        ServiceOutcome::Acted { receipt } => serde_json::json!({
            "outcome": "acted",
            "grant": receipt.grant,
        }),
        ServiceOutcome::Held { mode, grant, draft } => serde_json::json!({
            "outcome": "held",
            "mode": mode,
            "grant": grant,
            "draft": draft,
        }),
        ServiceOutcome::Policy(policy) => serde_json::json!({
            "outcome": "policy",
            "policy": policy,
        }),
    };
    body.to_string().into_bytes()
}