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};
pub(super) const SERVICE_REFUSED_CODE: u16 = 0x0301;
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()),
}
}
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))
}
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()
}