1use crate::convert::{ProtoPayload, ProtoRunId, ProtoWorkflowId, WireEnvelope};
4use crate::error::ProtoWireError;
5
6#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
8pub struct ProtoStartWorkflowRequest {
9 #[prost(string, tag = "1")]
11 pub namespace: String,
12 #[prost(string, tag = "2")]
14 pub workflow_type: String,
15 #[prost(message, optional, tag = "3")]
17 pub input: Option<ProtoPayload>,
18 #[prost(string, optional, tag = "4")]
22 pub routing_key: Option<String>,
23 #[prost(string, optional, tag = "5")]
28 pub task_queue: Option<String>,
29 #[prost(string, optional, tag = "6")]
42 pub display_name: Option<String>,
43}
44
45#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
47pub struct ProtoStartWorkflowResponse {
48 #[prost(message, optional, tag = "1")]
50 pub workflow_id: Option<ProtoWorkflowId>,
51 #[prost(message, optional, tag = "2")]
53 pub run_id: Option<ProtoRunId>,
54}
55
56#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
58pub struct ProtoSignalRequest {
59 #[prost(string, tag = "1")]
61 pub namespace: String,
62 #[prost(message, optional, tag = "2")]
64 pub workflow_id: Option<ProtoWorkflowId>,
65 #[prost(message, optional, tag = "3")]
67 pub run_id: Option<ProtoRunId>,
68 #[prost(string, tag = "4")]
70 pub signal_name: String,
71 #[prost(message, optional, tag = "5")]
73 pub payload: Option<ProtoPayload>,
74}
75
76#[derive(Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
78pub struct ProtoSignalResponse {}
79
80#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
82pub struct ProtoQueryRequest {
83 #[prost(string, tag = "1")]
85 pub namespace: String,
86 #[prost(message, optional, tag = "2")]
88 pub workflow_id: Option<ProtoWorkflowId>,
89 #[prost(message, optional, tag = "3")]
91 pub run_id: Option<ProtoRunId>,
92 #[prost(string, tag = "4")]
94 pub query_name: String,
95 #[prost(message, optional, tag = "5")]
101 pub arguments: Option<ProtoPayload>,
102}
103
104#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
106pub struct ProtoQueryResponse {
107 #[prost(oneof = "proto_query_response::Outcome", tags = "1, 2")]
109 pub outcome: Option<proto_query_response::Outcome>,
110}
111
112pub mod proto_query_response {
114 #[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Oneof)]
116 pub enum Outcome {
117 #[prost(message, tag = "1")]
119 Result(super::ProtoPayload),
120 #[prost(message, tag = "2")]
122 Error(super::ProtoWireError),
123 }
124}
125
126#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
128pub struct ProtoCancelRequest {
129 #[prost(string, tag = "1")]
131 pub namespace: String,
132 #[prost(message, optional, tag = "2")]
134 pub workflow_id: Option<ProtoWorkflowId>,
135 #[prost(message, optional, tag = "3")]
137 pub run_id: Option<ProtoRunId>,
138 #[prost(string, tag = "4")]
140 pub reason: String,
141}
142
143#[derive(Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
145pub struct ProtoCancelResponse {}
146
147#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
157pub struct ProtoRetireWorkloopRequest {
158 #[prost(string, tag = "1")]
160 pub namespace: String,
161 #[prost(message, optional, tag = "2")]
163 pub workflow_id: Option<ProtoWorkflowId>,
164 #[prost(string, tag = "3")]
166 pub reason: String,
167}
168
169#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
173pub struct ProtoRetireWorkloopResponse {
174 #[prost(string, tag = "1")]
176 pub reason: String,
177}
178
179#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
184pub struct ProtoReopenRequest {
185 #[prost(string, tag = "1")]
187 pub namespace: String,
188 #[prost(message, optional, tag = "2")]
190 pub workflow_id: Option<ProtoWorkflowId>,
191 #[prost(message, optional, tag = "3")]
193 pub run_id: Option<ProtoRunId>,
194}
195
196#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
202pub struct ProtoReopenResponse {
203 #[prost(message, optional, tag = "1")]
205 pub run_id: Option<ProtoRunId>,
206 #[prost(enumeration = "crate::convert::ProtoWorkflowStatus", tag = "2")]
208 pub status: i32,
209}
210
211#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
216pub struct ProtoPauseRequest {
217 #[prost(string, tag = "1")]
219 pub namespace: String,
220 #[prost(message, optional, tag = "2")]
222 pub workflow_id: Option<ProtoWorkflowId>,
223 #[prost(message, optional, tag = "3")]
225 pub run_id: Option<ProtoRunId>,
226 #[prost(string, tag = "4")]
228 pub reason: String,
229}
230
231#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
233pub struct ProtoPauseResponse {
234 #[prost(message, optional, tag = "1")]
236 pub run_id: Option<ProtoRunId>,
237 #[prost(enumeration = "crate::convert::ProtoWorkflowStatus", tag = "2")]
239 pub status: i32,
240}
241
242#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
247pub struct ProtoResumeRequest {
248 #[prost(string, tag = "1")]
250 pub namespace: String,
251 #[prost(message, optional, tag = "2")]
253 pub workflow_id: Option<ProtoWorkflowId>,
254 #[prost(message, optional, tag = "3")]
256 pub run_id: Option<ProtoRunId>,
257}
258
259#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
261pub struct ProtoResumeResponse {
262 #[prost(message, optional, tag = "1")]
264 pub run_id: Option<ProtoRunId>,
265 #[prost(enumeration = "crate::convert::ProtoWorkflowStatus", tag = "2")]
267 pub status: i32,
268}
269
270#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
284pub struct ProtoRenameRequest {
285 #[prost(string, tag = "1")]
287 pub namespace: String,
288 #[prost(message, optional, tag = "2")]
290 pub workflow_id: Option<ProtoWorkflowId>,
291 #[prost(message, optional, tag = "3")]
293 pub run_id: Option<ProtoRunId>,
294 #[prost(string, tag = "4")]
297 pub display_name: String,
298}
299
300#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
303pub struct ProtoRenameResponse {
304 #[prost(message, optional, tag = "1")]
306 pub run_id: Option<ProtoRunId>,
307 #[prost(string, tag = "2")]
309 pub display_name: String,
310}
311
312#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
314pub struct ProtoListWorkflowsRequest {
315 #[prost(string, tag = "1")]
317 pub namespace: String,
318 #[prost(message, optional, tag = "2")]
320 pub filter: Option<WireEnvelope>,
321}
322
323#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
325pub struct ProtoListWorkflowsResponse {
326 #[prost(message, repeated, tag = "1")]
328 pub summaries: Vec<WireEnvelope>,
329}
330
331#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
333pub struct ProtoCountWorkflowsRequest {
334 #[prost(string, tag = "1")]
336 pub namespace: String,
337 #[prost(message, optional, tag = "2")]
339 pub filter: Option<WireEnvelope>,
340}
341
342#[derive(Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
344pub struct ProtoCountWorkflowsResponse {
345 #[prost(uint64, tag = "1")]
347 pub count: u64,
348}
349
350#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
352pub struct ProtoDescribeWorkflowRequest {
353 #[prost(string, tag = "1")]
355 pub namespace: String,
356 #[prost(message, optional, tag = "2")]
358 pub workflow_id: Option<ProtoWorkflowId>,
359 #[prost(message, optional, tag = "3")]
361 pub run_id: Option<ProtoRunId>,
362 #[prost(bool, tag = "4")]
364 pub include_history: bool,
365}
366
367#[cfg(test)]
368mod tests {
369 use std::collections::HashMap;
370
371 use aion_core::SearchAttributeValue;
372 use aion_store::visibility::{ListWorkflowsFilter, SearchAttributePredicate};
373 use chrono::{DateTime, Utc};
374 use prost::Message;
375 use serde::de::DeserializeOwned;
376 use serde_json::json;
377
378 use super::{
379 ProtoCountWorkflowsRequest, ProtoCountWorkflowsResponse, ProtoListWorkflowsRequest,
380 ProtoListWorkflowsResponse, ProtoQueryRequest, ProtoQueryResponse, ProtoReopenRequest,
381 ProtoReopenResponse, ProtoStartWorkflowRequest, ProtoStartWorkflowResponse,
382 proto_query_response,
383 };
384 use crate::convert::{
385 ProtoPayload, ProtoRunId, ProtoWorkflowId, decode_core_value, encode_core_value,
386 };
387 use crate::error::{ProtoWireError, WireError};
388
389 fn workflow_id() -> aion_core::WorkflowId {
390 aion_core::WorkflowId::new(uuid::Uuid::nil())
391 }
392
393 fn run_id() -> aion_core::RunId {
394 aion_core::RunId::new(uuid::Uuid::nil())
395 }
396
397 fn payload(label: &str) -> Result<ProtoPayload, aion_core::PayloadError> {
398 Ok(ProtoPayload::from(aion_core::Payload::from_json(
399 &json!({ "label": label }),
400 )?))
401 }
402
403 fn recorded_at() -> Result<DateTime<Utc>, chrono::ParseError> {
404 Ok(DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")?.with_timezone(&Utc))
405 }
406
407 fn assert_json_round_trip<T>(value: &T) -> Result<(), serde_json::Error>
408 where
409 T: Clone + PartialEq + serde::Serialize + DeserializeOwned,
410 {
411 let encoded = serde_json::to_string(value)?;
412 let decoded = serde_json::from_str::<T>(&encoded)?;
413 assert!(decoded == *value);
414 Ok(())
415 }
416
417 fn assert_proto_round_trip<T>(value: &T) -> Result<(), Box<dyn std::error::Error>>
418 where
419 T: Clone + PartialEq + Message + Default,
420 {
421 let mut bytes = Vec::new();
422 value.encode(&mut bytes)?;
423 let decoded = T::decode(bytes.as_slice())?;
424 assert!(decoded == *value);
425 Ok(())
426 }
427
428 #[test]
429 fn start_workflow_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
430 let request = ProtoStartWorkflowRequest {
431 namespace: String::from("tenant-a"),
432 workflow_type: String::from("checkout"),
433 input: Some(payload("input")?),
434 routing_key: Some(String::from("tenant-a/order-1")),
435 task_queue: Some(String::from("gpu")),
436 display_name: Some(String::from("Order 1 checkout")),
437 };
438 let response = ProtoStartWorkflowResponse {
439 workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
440 run_id: Some(ProtoRunId::from(run_id())),
441 };
442
443 assert_json_round_trip(&request)?;
444 assert_proto_round_trip(&request)?;
445 assert_json_round_trip(&response)?;
446 assert_proto_round_trip(&response)?;
447 Ok(())
448 }
449
450 #[test]
451 fn list_workflows_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
452 let filter = ListWorkflowsFilter {
453 workflow_type: Some(String::from("checkout")),
454 status: Some(aion_core::WorkflowStatus::Running),
455 search_attributes: vec![SearchAttributePredicate::Equals {
456 name: String::from("customer_id"),
457 value: SearchAttributeValue::String(String::from("12345")),
458 }],
459 limit: Some(10),
460 offset: Some(5),
461 ..ListWorkflowsFilter::default()
462 };
463 let summary = aion_store::visibility::WorkflowSummary {
464 workflow_id: workflow_id(),
465 run_id: run_id(),
466 workflow_type: String::from("checkout"),
467 status: aion_core::WorkflowStatus::Running,
468 start_time: recorded_at()?,
469 close_time: None,
470 failed_step: None,
471 failure_reason: None,
472 search_attributes: HashMap::from([(
473 String::from("customer_id"),
474 SearchAttributeValue::String(String::from("12345")),
475 )]),
476 };
477 let filter_envelope = encode_core_value("tenant-a", Some(String::from("r1")), &filter)?;
478 let summary_envelope = encode_core_value("tenant-a", None, &summary)?;
479 let request = ProtoListWorkflowsRequest {
480 namespace: String::from("tenant-a"),
481 filter: Some(filter_envelope.clone()),
482 };
483 let response = ProtoListWorkflowsResponse {
484 summaries: vec![summary_envelope.clone()],
485 };
486 let count_request = ProtoCountWorkflowsRequest {
487 namespace: String::from("tenant-a"),
488 filter: Some(filter_envelope.clone()),
489 };
490 let count_response = ProtoCountWorkflowsResponse { count: 1 };
491
492 assert_json_round_trip(&request)?;
493 assert_proto_round_trip(&request)?;
494 assert_json_round_trip(&response)?;
495 assert_proto_round_trip(&response)?;
496 assert_json_round_trip(&count_request)?;
497 assert_proto_round_trip(&count_request)?;
498 assert_json_round_trip(&count_response)?;
499 assert_proto_round_trip(&count_response)?;
500 assert_eq!(
501 decode_core_value::<ListWorkflowsFilter>(&filter_envelope)?,
502 filter
503 );
504 assert_eq!(
505 decode_core_value::<aion_store::visibility::WorkflowSummary>(&summary_envelope)?,
506 summary
507 );
508 Ok(())
509 }
510
511 #[test]
512 fn query_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
513 let request = ProtoQueryRequest {
514 namespace: String::from("tenant-a"),
515 workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
516 run_id: Some(ProtoRunId::from(run_id())),
517 query_name: String::from("state"),
518 arguments: Some(payload("arguments")?),
519 };
520 let no_arguments_request = ProtoQueryRequest {
524 arguments: None,
525 ..request.clone()
526 };
527 let result_response = ProtoQueryResponse {
528 outcome: Some(proto_query_response::Outcome::Result(payload("result")?)),
529 };
530 let error_response = ProtoQueryResponse {
531 outcome: Some(proto_query_response::Outcome::Error(ProtoWireError::from(
532 WireError::unknown_query("state query is not registered"),
533 ))),
534 };
535
536 assert_json_round_trip(&request)?;
537 assert_proto_round_trip(&request)?;
538 assert_json_round_trip(&no_arguments_request)?;
539 assert_proto_round_trip(&no_arguments_request)?;
540 assert_json_round_trip(&result_response)?;
541 assert_proto_round_trip(&result_response)?;
542 assert_json_round_trip(&error_response)?;
543 assert_proto_round_trip(&error_response)?;
544 assert_ne!(request, no_arguments_request);
547 Ok(())
548 }
549
550 #[test]
553 fn start_workflow_display_name_absent_round_trips() -> Result<(), Box<dyn std::error::Error>> {
554 let named = ProtoStartWorkflowRequest {
555 namespace: String::from("tenant-a"),
556 workflow_type: String::from("checkout"),
557 input: Some(payload("input")?),
558 routing_key: None,
559 task_queue: None,
560 display_name: Some(String::from("Order 1 checkout")),
561 };
562 let unnamed = ProtoStartWorkflowRequest {
563 display_name: None,
564 ..named.clone()
565 };
566
567 assert_json_round_trip(&named)?;
568 assert_proto_round_trip(&named)?;
569 assert_json_round_trip(&unnamed)?;
570 assert_proto_round_trip(&unnamed)?;
571 assert_ne!(named, unnamed);
574 Ok(())
575 }
576
577 #[cfg(feature = "generated")]
587 #[test]
588 fn start_workflow_display_name_is_the_same_wire_field_as_the_generated_stub()
589 -> Result<(), Box<dyn std::error::Error>> {
590 const NAME: &str = "Nightly settlement";
591 let hand_written = ProtoStartWorkflowRequest {
592 namespace: String::from("tenant-a"),
593 workflow_type: String::from("checkout"),
594 input: None,
595 routing_key: None,
596 task_queue: None,
597 display_name: Some(String::from(NAME)),
598 };
599
600 let mut bytes = Vec::new();
602 hand_written.encode(&mut bytes)?;
603 let decoded = crate::generated::StartWorkflowRequest::decode(bytes.as_slice())?;
604 assert_eq!(
605 decoded.display_name.as_deref(),
606 Some(NAME),
607 "the generated stub must read the hand-written display_name"
608 );
609
610 let mut bytes = Vec::new();
612 decoded.encode(&mut bytes)?;
613 let round_tripped = ProtoStartWorkflowRequest::decode(bytes.as_slice())?;
614 assert_eq!(round_tripped, hand_written);
615
616 let mut bytes = Vec::new();
619 ProtoStartWorkflowRequest {
620 namespace: String::new(),
621 workflow_type: String::new(),
622 input: None,
623 routing_key: None,
624 task_queue: None,
625 display_name: Some(String::from("x")),
626 }
627 .encode(&mut bytes)?;
628 assert_eq!(bytes, vec![0x32, 0x01, b'x']);
629
630 let mut bytes = Vec::new();
631 ProtoStartWorkflowRequest {
632 namespace: String::new(),
633 workflow_type: String::new(),
634 input: None,
635 routing_key: None,
636 task_queue: None,
637 display_name: None,
638 }
639 .encode(&mut bytes)?;
640 assert!(
641 bytes.is_empty(),
642 "an unnamed start must put nothing on the wire, got {bytes:?}"
643 );
644 Ok(())
645 }
646
647 #[cfg(feature = "generated")]
650 #[test]
651 fn rename_messages_agree_with_the_generated_stubs() -> Result<(), Box<dyn std::error::Error>> {
652 let request = super::ProtoRenameRequest {
653 namespace: String::from("tenant-a"),
654 workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
655 run_id: Some(ProtoRunId::from(run_id())),
656 display_name: String::from("Nightly settlement"),
657 };
658 let mut bytes = Vec::new();
659 request.encode(&mut bytes)?;
660 let decoded = crate::generated::RenameRequest::decode(bytes.as_slice())?;
661 assert_eq!(decoded.display_name, "Nightly settlement");
662 assert_eq!(decoded.namespace, "tenant-a");
663 let mut bytes = Vec::new();
664 decoded.encode(&mut bytes)?;
665 assert_eq!(
666 super::ProtoRenameRequest::decode(bytes.as_slice())?,
667 request
668 );
669
670 let response = super::ProtoRenameResponse {
671 run_id: Some(ProtoRunId::from(run_id())),
672 display_name: String::from("Nightly settlement"),
673 };
674 let mut bytes = Vec::new();
675 response.encode(&mut bytes)?;
676 let decoded = crate::generated::RenameResponse::decode(bytes.as_slice())?;
677 assert_eq!(decoded.display_name, "Nightly settlement");
678 let mut bytes = Vec::new();
679 decoded.encode(&mut bytes)?;
680 assert_eq!(
681 super::ProtoRenameResponse::decode(bytes.as_slice())?,
682 response
683 );
684 Ok(())
685 }
686
687 #[test]
688 fn rename_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
689 let request = super::ProtoRenameRequest {
690 namespace: String::from("tenant-a"),
691 workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
692 run_id: Some(ProtoRunId::from(run_id())),
693 display_name: String::from("Nightly settlement"),
694 };
695 let response = super::ProtoRenameResponse {
696 run_id: Some(ProtoRunId::from(run_id())),
697 display_name: String::from("Nightly settlement"),
698 };
699
700 assert_json_round_trip(&request)?;
701 assert_proto_round_trip(&request)?;
702 assert_json_round_trip(&response)?;
703 assert_proto_round_trip(&response)?;
704 Ok(())
705 }
706
707 #[test]
708 fn reopen_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
709 let request = ProtoReopenRequest {
710 namespace: String::from("tenant-a"),
711 workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
712 run_id: Some(ProtoRunId::from(run_id())),
713 };
714 let response = ProtoReopenResponse {
715 run_id: Some(ProtoRunId::from(run_id())),
716 status: crate::convert::ProtoWorkflowStatus::Running as i32,
717 };
718
719 assert_json_round_trip(&request)?;
720 assert_proto_round_trip(&request)?;
721 assert_json_round_trip(&response)?;
722 assert_proto_round_trip(&response)?;
723 Ok(())
724 }
725}