use crate::convert::{ProtoPayload, ProtoRunId, ProtoWorkflowId, WireEnvelope};
use crate::error::ProtoWireError;
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoStartWorkflowRequest {
#[prost(string, tag = "1")]
pub namespace: String,
#[prost(string, tag = "2")]
pub workflow_type: String,
#[prost(message, optional, tag = "3")]
pub input: Option<ProtoPayload>,
#[prost(string, optional, tag = "4")]
pub routing_key: Option<String>,
#[prost(string, optional, tag = "5")]
pub task_queue: Option<String>,
#[prost(string, optional, tag = "6")]
pub display_name: Option<String>,
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoStartWorkflowResponse {
#[prost(message, optional, tag = "1")]
pub workflow_id: Option<ProtoWorkflowId>,
#[prost(message, optional, tag = "2")]
pub run_id: Option<ProtoRunId>,
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoSignalRequest {
#[prost(string, tag = "1")]
pub namespace: String,
#[prost(message, optional, tag = "2")]
pub workflow_id: Option<ProtoWorkflowId>,
#[prost(message, optional, tag = "3")]
pub run_id: Option<ProtoRunId>,
#[prost(string, tag = "4")]
pub signal_name: String,
#[prost(message, optional, tag = "5")]
pub payload: Option<ProtoPayload>,
}
#[derive(Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoSignalResponse {}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoQueryRequest {
#[prost(string, tag = "1")]
pub namespace: String,
#[prost(message, optional, tag = "2")]
pub workflow_id: Option<ProtoWorkflowId>,
#[prost(message, optional, tag = "3")]
pub run_id: Option<ProtoRunId>,
#[prost(string, tag = "4")]
pub query_name: String,
#[prost(message, optional, tag = "5")]
pub arguments: Option<ProtoPayload>,
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoQueryResponse {
#[prost(oneof = "proto_query_response::Outcome", tags = "1, 2")]
pub outcome: Option<proto_query_response::Outcome>,
}
pub mod proto_query_response {
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Oneof)]
pub enum Outcome {
#[prost(message, tag = "1")]
Result(super::ProtoPayload),
#[prost(message, tag = "2")]
Error(super::ProtoWireError),
}
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoCancelRequest {
#[prost(string, tag = "1")]
pub namespace: String,
#[prost(message, optional, tag = "2")]
pub workflow_id: Option<ProtoWorkflowId>,
#[prost(message, optional, tag = "3")]
pub run_id: Option<ProtoRunId>,
#[prost(string, tag = "4")]
pub reason: String,
}
#[derive(Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoCancelResponse {}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoRetireWorkloopRequest {
#[prost(string, tag = "1")]
pub namespace: String,
#[prost(message, optional, tag = "2")]
pub workflow_id: Option<ProtoWorkflowId>,
#[prost(string, tag = "3")]
pub reason: String,
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoRetireWorkloopResponse {
#[prost(string, tag = "1")]
pub reason: String,
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoReopenRequest {
#[prost(string, tag = "1")]
pub namespace: String,
#[prost(message, optional, tag = "2")]
pub workflow_id: Option<ProtoWorkflowId>,
#[prost(message, optional, tag = "3")]
pub run_id: Option<ProtoRunId>,
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoReopenResponse {
#[prost(message, optional, tag = "1")]
pub run_id: Option<ProtoRunId>,
#[prost(enumeration = "crate::convert::ProtoWorkflowStatus", tag = "2")]
pub status: i32,
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoPauseRequest {
#[prost(string, tag = "1")]
pub namespace: String,
#[prost(message, optional, tag = "2")]
pub workflow_id: Option<ProtoWorkflowId>,
#[prost(message, optional, tag = "3")]
pub run_id: Option<ProtoRunId>,
#[prost(string, tag = "4")]
pub reason: String,
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoPauseResponse {
#[prost(message, optional, tag = "1")]
pub run_id: Option<ProtoRunId>,
#[prost(enumeration = "crate::convert::ProtoWorkflowStatus", tag = "2")]
pub status: i32,
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoResumeRequest {
#[prost(string, tag = "1")]
pub namespace: String,
#[prost(message, optional, tag = "2")]
pub workflow_id: Option<ProtoWorkflowId>,
#[prost(message, optional, tag = "3")]
pub run_id: Option<ProtoRunId>,
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoResumeResponse {
#[prost(message, optional, tag = "1")]
pub run_id: Option<ProtoRunId>,
#[prost(enumeration = "crate::convert::ProtoWorkflowStatus", tag = "2")]
pub status: i32,
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoRenameRequest {
#[prost(string, tag = "1")]
pub namespace: String,
#[prost(message, optional, tag = "2")]
pub workflow_id: Option<ProtoWorkflowId>,
#[prost(message, optional, tag = "3")]
pub run_id: Option<ProtoRunId>,
#[prost(string, tag = "4")]
pub display_name: String,
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoRenameResponse {
#[prost(message, optional, tag = "1")]
pub run_id: Option<ProtoRunId>,
#[prost(string, tag = "2")]
pub display_name: String,
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoListWorkflowsRequest {
#[prost(string, tag = "1")]
pub namespace: String,
#[prost(message, optional, tag = "2")]
pub request: Option<WireEnvelope>,
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoListWorkflowsResponse {
#[prost(message, optional, tag = "1")]
pub page: Option<WireEnvelope>,
}
#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
pub struct ProtoDescribeWorkflowRequest {
#[prost(string, tag = "1")]
pub namespace: String,
#[prost(message, optional, tag = "2")]
pub workflow_id: Option<ProtoWorkflowId>,
#[prost(message, optional, tag = "3")]
pub run_id: Option<ProtoRunId>,
#[prost(bool, tag = "4")]
pub include_history: bool,
}
#[cfg(test)]
mod tests {
use aion_core::{
SortDirection, WorkflowListFilter, WorkflowListPage, WorkflowListRequest, WorkflowSort,
WorkflowSortField, WorkflowSummary,
};
use chrono::{DateTime, Utc};
use prost::Message;
use serde::de::DeserializeOwned;
use serde_json::json;
use super::{
ProtoListWorkflowsRequest, ProtoListWorkflowsResponse, ProtoQueryRequest,
ProtoQueryResponse, ProtoReopenRequest, ProtoReopenResponse, ProtoStartWorkflowRequest,
ProtoStartWorkflowResponse, proto_query_response,
};
use crate::convert::{
ProtoPayload, ProtoRunId, ProtoWorkflowId, decode_core_value, encode_core_value,
};
use crate::error::{ProtoWireError, WireError};
fn workflow_id() -> aion_core::WorkflowId {
aion_core::WorkflowId::new(uuid::Uuid::nil())
}
fn run_id() -> aion_core::RunId {
aion_core::RunId::new(uuid::Uuid::nil())
}
fn payload(label: &str) -> Result<ProtoPayload, aion_core::PayloadError> {
Ok(ProtoPayload::from(aion_core::Payload::from_json(
&json!({ "label": label }),
)?))
}
fn recorded_at() -> Result<DateTime<Utc>, chrono::ParseError> {
Ok(DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")?.with_timezone(&Utc))
}
fn assert_json_round_trip<T>(value: &T) -> Result<(), serde_json::Error>
where
T: Clone + PartialEq + serde::Serialize + DeserializeOwned,
{
let encoded = serde_json::to_string(value)?;
let decoded = serde_json::from_str::<T>(&encoded)?;
assert!(decoded == *value);
Ok(())
}
fn assert_proto_round_trip<T>(value: &T) -> Result<(), Box<dyn std::error::Error>>
where
T: Clone + PartialEq + Message + Default,
{
let mut bytes = Vec::new();
value.encode(&mut bytes)?;
let decoded = T::decode(bytes.as_slice())?;
assert!(decoded == *value);
Ok(())
}
#[test]
fn start_workflow_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
let request = ProtoStartWorkflowRequest {
namespace: String::from("tenant-a"),
workflow_type: String::from("checkout"),
input: Some(payload("input")?),
routing_key: Some(String::from("tenant-a/order-1")),
task_queue: Some(String::from("gpu")),
display_name: Some(String::from("Order 1 checkout")),
};
let response = ProtoStartWorkflowResponse {
workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
run_id: Some(ProtoRunId::from(run_id())),
};
assert_json_round_trip(&request)?;
assert_proto_round_trip(&request)?;
assert_json_round_trip(&response)?;
assert_proto_round_trip(&response)?;
Ok(())
}
#[test]
fn list_workflows_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
let list_request = WorkflowListRequest {
namespace: String::from("tenant-a"),
filter: WorkflowListFilter {
workflow_types: vec![String::from("checkout")],
statuses: vec![aion_core::WorkflowStatus::Running],
..WorkflowListFilter::default()
},
sort: WorkflowSort {
field: WorkflowSortField::UpdatedAt,
direction: SortDirection::Desc,
},
cursor: Some(String::from("opaque")),
limit: 10,
};
let page = WorkflowListPage {
items: vec![WorkflowSummary {
workflow_id: workflow_id(),
run_id: run_id(),
workflow_type: String::from("checkout"),
status: aion_core::WorkflowStatus::Running,
started_at: recorded_at()?,
updated_at: recorded_at()?,
ended_at: None,
parent: None,
failed_step: None,
failure_reason: None,
display_name: Some(String::from("Nightly close")),
kind: None,
current_worker: None,
package_version: None,
}],
next_cursor: Some(String::from("next")),
count: 7,
provenance: None,
};
let request_envelope =
encode_core_value("tenant-a", Some(String::from("r1")), &list_request)?;
let page_envelope = encode_core_value("tenant-a", None, &page)?;
let request = ProtoListWorkflowsRequest {
namespace: String::from("tenant-a"),
request: Some(request_envelope.clone()),
};
let response = ProtoListWorkflowsResponse {
page: Some(page_envelope.clone()),
};
assert_json_round_trip(&request)?;
assert_proto_round_trip(&request)?;
assert_json_round_trip(&response)?;
assert_proto_round_trip(&response)?;
assert_eq!(
decode_core_value::<WorkflowListRequest>(&request_envelope)?,
list_request
);
assert_eq!(decode_core_value::<WorkflowListPage>(&page_envelope)?, page);
Ok(())
}
#[test]
fn query_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
let request = ProtoQueryRequest {
namespace: String::from("tenant-a"),
workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
run_id: Some(ProtoRunId::from(run_id())),
query_name: String::from("state"),
arguments: Some(payload("arguments")?),
};
let no_arguments_request = ProtoQueryRequest {
arguments: None,
..request.clone()
};
let result_response = ProtoQueryResponse {
outcome: Some(proto_query_response::Outcome::Result(payload("result")?)),
};
let error_response = ProtoQueryResponse {
outcome: Some(proto_query_response::Outcome::Error(ProtoWireError::from(
WireError::unknown_query("state query is not registered"),
))),
};
assert_json_round_trip(&request)?;
assert_proto_round_trip(&request)?;
assert_json_round_trip(&no_arguments_request)?;
assert_proto_round_trip(&no_arguments_request)?;
assert_json_round_trip(&result_response)?;
assert_proto_round_trip(&result_response)?;
assert_json_round_trip(&error_response)?;
assert_proto_round_trip(&error_response)?;
assert_ne!(request, no_arguments_request);
Ok(())
}
#[test]
fn start_workflow_display_name_absent_round_trips() -> Result<(), Box<dyn std::error::Error>> {
let named = ProtoStartWorkflowRequest {
namespace: String::from("tenant-a"),
workflow_type: String::from("checkout"),
input: Some(payload("input")?),
routing_key: None,
task_queue: None,
display_name: Some(String::from("Order 1 checkout")),
};
let unnamed = ProtoStartWorkflowRequest {
display_name: None,
..named.clone()
};
assert_json_round_trip(&named)?;
assert_proto_round_trip(&named)?;
assert_json_round_trip(&unnamed)?;
assert_proto_round_trip(&unnamed)?;
assert_ne!(named, unnamed);
Ok(())
}
#[cfg(feature = "generated")]
#[test]
fn start_workflow_display_name_is_the_same_wire_field_as_the_generated_stub()
-> Result<(), Box<dyn std::error::Error>> {
const NAME: &str = "Nightly settlement";
let hand_written = ProtoStartWorkflowRequest {
namespace: String::from("tenant-a"),
workflow_type: String::from("checkout"),
input: None,
routing_key: None,
task_queue: None,
display_name: Some(String::from(NAME)),
};
let mut bytes = Vec::new();
hand_written.encode(&mut bytes)?;
let decoded = crate::generated::StartWorkflowRequest::decode(bytes.as_slice())?;
assert_eq!(
decoded.display_name.as_deref(),
Some(NAME),
"the generated stub must read the hand-written display_name"
);
let mut bytes = Vec::new();
decoded.encode(&mut bytes)?;
let round_tripped = ProtoStartWorkflowRequest::decode(bytes.as_slice())?;
assert_eq!(round_tripped, hand_written);
let mut bytes = Vec::new();
ProtoStartWorkflowRequest {
namespace: String::new(),
workflow_type: String::new(),
input: None,
routing_key: None,
task_queue: None,
display_name: Some(String::from("x")),
}
.encode(&mut bytes)?;
assert_eq!(bytes, vec![0x32, 0x01, b'x']);
let mut bytes = Vec::new();
ProtoStartWorkflowRequest {
namespace: String::new(),
workflow_type: String::new(),
input: None,
routing_key: None,
task_queue: None,
display_name: None,
}
.encode(&mut bytes)?;
assert!(
bytes.is_empty(),
"an unnamed start must put nothing on the wire, got {bytes:?}"
);
Ok(())
}
#[cfg(feature = "generated")]
#[test]
fn rename_messages_agree_with_the_generated_stubs() -> Result<(), Box<dyn std::error::Error>> {
let request = super::ProtoRenameRequest {
namespace: String::from("tenant-a"),
workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
run_id: Some(ProtoRunId::from(run_id())),
display_name: String::from("Nightly settlement"),
};
let mut bytes = Vec::new();
request.encode(&mut bytes)?;
let decoded = crate::generated::RenameRequest::decode(bytes.as_slice())?;
assert_eq!(decoded.display_name, "Nightly settlement");
assert_eq!(decoded.namespace, "tenant-a");
let mut bytes = Vec::new();
decoded.encode(&mut bytes)?;
assert_eq!(
super::ProtoRenameRequest::decode(bytes.as_slice())?,
request
);
let response = super::ProtoRenameResponse {
run_id: Some(ProtoRunId::from(run_id())),
display_name: String::from("Nightly settlement"),
};
let mut bytes = Vec::new();
response.encode(&mut bytes)?;
let decoded = crate::generated::RenameResponse::decode(bytes.as_slice())?;
assert_eq!(decoded.display_name, "Nightly settlement");
let mut bytes = Vec::new();
decoded.encode(&mut bytes)?;
assert_eq!(
super::ProtoRenameResponse::decode(bytes.as_slice())?,
response
);
Ok(())
}
#[test]
fn rename_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
let request = super::ProtoRenameRequest {
namespace: String::from("tenant-a"),
workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
run_id: Some(ProtoRunId::from(run_id())),
display_name: String::from("Nightly settlement"),
};
let response = super::ProtoRenameResponse {
run_id: Some(ProtoRunId::from(run_id())),
display_name: String::from("Nightly settlement"),
};
assert_json_round_trip(&request)?;
assert_proto_round_trip(&request)?;
assert_json_round_trip(&response)?;
assert_proto_round_trip(&response)?;
Ok(())
}
#[test]
fn reopen_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
let request = ProtoReopenRequest {
namespace: String::from("tenant-a"),
workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
run_id: Some(ProtoRunId::from(run_id())),
};
let response = ProtoReopenResponse {
run_id: Some(ProtoRunId::from(run_id())),
status: crate::convert::ProtoWorkflowStatus::Running as i32,
};
assert_json_round_trip(&request)?;
assert_proto_round_trip(&request)?;
assert_json_round_trip(&response)?;
assert_proto_round_trip(&response)?;
Ok(())
}
}