use std::io::{self, BufRead, Write};
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use serde_json::{Value, json};
use crate::store::StrategyStore;
use crate::{BindingValue, ReducerEvent, WorkflowDefinition};
pub const PROTOCOL: &str = "acativa.runtime.v1";
pub const OPERATIONS: &[&str] = &[
"describe",
"admit",
"snapshot",
"lookup",
"definition",
"claim",
"event",
"complete",
"receipt",
"authorize",
"renew",
"recover",
"prune",
];
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct Request {
pub protocol: String,
pub id: u64,
pub operation: String,
#[serde(deserialize_with = "Value::deserialize")]
pub payload: Value,
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Response {
pub protocol: &'static str,
pub id: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub result: Option<Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<ProtocolError>,
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ProtocolError {
pub code: &'static str,
#[serde(skip_serializing_if = "Option::is_none")]
pub reason: Option<&'static str>,
}
impl ProtocolError {
fn new(code: &'static str) -> Self {
Self { code, reason: None }
}
}
impl From<anyhow::Error> for ProtocolError {
fn from(error: anyhow::Error) -> Self {
let reason = match error.to_string().as_str() {
"strategy_idempotency_conflict" => Some("strategy_idempotency_conflict"),
"strategy_callback_stale" => Some("strategy_callback_stale"),
"strategy_callback_conflict" => Some("strategy_callback_conflict"),
"strategy_lease_lost" => Some("strategy_lease_lost"),
"strategy_authorization_stale" => Some("strategy_authorization_stale"),
"strategy_authorization_required" => Some("strategy_authorization_required"),
"strategy_receipt_conflict" => Some("strategy_receipt_conflict"),
"strategy_receipt_missing" => Some("strategy_receipt_missing"),
"strategy_receipt_event_invalid" => Some("strategy_receipt_event_invalid"),
"strategy_receipt_too_large" => Some("strategy_receipt_too_large"),
"strategy_prune_bounds_invalid" => Some("strategy_prune_bounds_invalid"),
"strategy_run_not_found" => Some("strategy_run_not_found"),
"strategy_definition_not_found" => Some("strategy_definition_not_found"),
_ => None,
};
Self {
code: "runtime_refused",
reason,
}
}
}
impl Response {
fn from_result(id: Option<u64>, result: Result<Value, ProtocolError>) -> Self {
let (result, error) = match result {
Ok(value) => (Some(value), None),
Err(error) => (None, Some(error)),
};
Self {
protocol: PROTOCOL,
id,
result,
error,
}
}
fn invalid_request() -> Self {
Self::from_result(None, Err(ProtocolError::new("invalid_request")))
}
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct Describe {}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct Admit {
revision_digest: String,
semantics_digest: String,
workflow: WorkflowDefinition,
bindings: Vec<BindingValue>,
#[serde(deserialize_with = "Value::deserialize")]
input: Value,
idempotency_key: String,
context_id: String,
#[serde(deserialize_with = "Value::deserialize")]
run_binding: Value,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct Run {
run_id: String,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct Lookup {
idempotency_key: String,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct Definition {
revision_digest: String,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct Claim {
run_id: String,
claimant: String,
lease_until_unix_ms: i64,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct Event {
run_id: String,
claimant: String,
event: ReducerEvent,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct Complete {
run_id: String,
claimant: String,
event: ReducerEvent,
#[serde(deserialize_with = "Value::deserialize")]
receipt: Value,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct Receipt {
run_id: String,
command_id: String,
attempt_token: String,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct Authorize {
run_id: String,
command_id: String,
attempt_token: String,
expected_authorization_digest: String,
claimant: String,
lease_until_unix_ms: i64,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct Renew {
command_id: String,
claimant: String,
lease_until_unix_ms: i64,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct Prune {
keep_terminal_runs: u64,
keep_events_per_run: u64,
vacuum: bool,
}
fn payload<T: DeserializeOwned>(value: Value) -> Result<T, ProtocolError> {
if !value.is_object() {
return Err(ProtocolError::new("invalid_payload"));
}
serde_json::from_value(value).map_err(|_| ProtocolError::new("invalid_payload"))
}
fn encoded(value: impl Serialize) -> Result<Value, ProtocolError> {
serde_json::to_value(value).map_err(|_| ProtocolError::new("runtime_refused"))
}
pub struct RuntimeProtocol {
store: StrategyStore,
}
impl RuntimeProtocol {
pub fn new(store: StrategyStore) -> Self {
Self { store }
}
pub fn handle(&self, request: Request) -> Response {
let result = if request.protocol == PROTOCOL {
self.dispatch(&request.operation, request.payload)
} else {
Err(ProtocolError::new("invalid_request"))
};
Response::from_result(Some(request.id), result)
}
fn dispatch(&self, operation: &str, value: Value) -> Result<Value, ProtocolError> {
match operation {
"describe" => {
let _: Describe = payload(value)?;
Ok(json!({
"protocol": PROTOCOL,
"version": env!("CARGO_PKG_VERSION"),
"workflowSchema": self.store.dialect().workflow_schema.as_str(),
"projectionSchema": self.store.dialect().projection_schema.as_str(),
"operations": OPERATIONS,
}))
}
"admit" => {
let request: Admit = payload(value)?;
let (snapshot, created) = self.store.admit_ephemeral_run(
&request.revision_digest,
&request.semantics_digest,
&request.workflow,
&request.bindings,
request.input,
&request.idempotency_key,
&request.context_id,
request.run_binding,
)?;
Ok(json!({ "snapshot": snapshot, "created": created }))
}
"snapshot" => {
let request: Run = payload(value)?;
encoded(self.store.run(&request.run_id)?)
}
"lookup" => {
let request: Lookup = payload(value)?;
encoded(
self.store
.run_id_by_idempotency_key(&request.idempotency_key)?,
)
}
"definition" => {
let request: Definition = payload(value)?;
encoded(
self.store
.definition_by_revision(&request.revision_digest)?,
)
}
"claim" => {
let request: Claim = payload(value)?;
encoded(self.store.claim_next_command(
&request.run_id,
&request.claimant,
request.lease_until_unix_ms,
)?)
}
"event" => {
let request: Event = payload(value)?;
encoded(self.store.apply_owned_event(
&request.run_id,
&request.claimant,
request.event,
)?)
}
"complete" => {
let request: Complete = payload(value)?;
encoded(self.store.apply_owned_event_with_receipt(
&request.run_id,
&request.claimant,
request.event,
request.receipt,
)?)
}
"receipt" => {
let request: Receipt = payload(value)?;
encoded(self.store.command_receipt(
&request.run_id,
&request.command_id,
&request.attempt_token,
)?)
}
"authorize" => {
let request: Authorize = payload(value)?;
self.store.authorize_effect(
&request.run_id,
&request.command_id,
&request.attempt_token,
&request.expected_authorization_digest,
&request.claimant,
request.lease_until_unix_ms,
)?;
Ok(Value::Null)
}
"renew" => {
let request: Renew = payload(value)?;
self.store.renew_command_lease(
&request.command_id,
&request.claimant,
request.lease_until_unix_ms,
)?;
Ok(Value::Null)
}
"recover" => {
let request: Run = payload(value)?;
while self.store.recover_next_expired_command(&request.run_id)? {}
Ok(json!({
"snapshot": self.store.run(&request.run_id)?,
"nextClaimExpiryUnixMs": self.store.next_claim_expiry(&request.run_id)?,
}))
}
"prune" => {
let request: Prune = payload(value)?;
encoded(self.store.prune(
request.keep_terminal_runs,
request.keep_events_per_run,
request.vacuum,
)?)
}
_ => Err(ProtocolError::new("unsupported_operation")),
}
}
}
pub fn serve(
mut reader: impl BufRead,
mut writer: impl Write,
store: StrategyStore,
) -> io::Result<()> {
let runtime = RuntimeProtocol::new(store);
let mut line = String::new();
loop {
line.clear();
let response = match reader.read_line(&mut line) {
Ok(0) => return Ok(()),
Ok(_) if line.trim_start().starts_with('{') => {
match serde_json::from_str::<Request>(&line) {
Ok(request) => runtime.handle(request),
Err(_) => Response::invalid_request(),
}
}
Ok(_) => Response::invalid_request(),
Err(error) if error.kind() == io::ErrorKind::InvalidData => Response::invalid_request(),
Err(error) => return Err(error),
};
serde_json::to_writer(&mut writer, &response)?;
writer.write_all(b"\n")?;
writer.flush()?;
}
}