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")]
318 pub namespace: String,
319 #[prost(message, optional, tag = "2")]
322 pub request: Option<WireEnvelope>,
323}
324
325#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
327pub struct ProtoListWorkflowsResponse {
328 #[prost(message, optional, tag = "1")]
331 pub page: Option<WireEnvelope>,
332}
333
334#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, prost::Message)]
336pub struct ProtoDescribeWorkflowRequest {
337 #[prost(string, tag = "1")]
339 pub namespace: String,
340 #[prost(message, optional, tag = "2")]
342 pub workflow_id: Option<ProtoWorkflowId>,
343 #[prost(message, optional, tag = "3")]
345 pub run_id: Option<ProtoRunId>,
346 #[prost(bool, tag = "4")]
348 pub include_history: bool,
349}
350
351#[cfg(test)]
352mod tests {
353 use aion_core::{
354 SortDirection, WorkflowListFilter, WorkflowListPage, WorkflowListRequest, WorkflowSort,
355 WorkflowSortField, WorkflowSummary,
356 };
357 use chrono::{DateTime, Utc};
358 use prost::Message;
359 use serde::de::DeserializeOwned;
360 use serde_json::json;
361
362 use super::{
363 ProtoListWorkflowsRequest, ProtoListWorkflowsResponse, ProtoQueryRequest,
364 ProtoQueryResponse, ProtoReopenRequest, ProtoReopenResponse, ProtoStartWorkflowRequest,
365 ProtoStartWorkflowResponse, proto_query_response,
366 };
367 use crate::convert::{
368 ProtoPayload, ProtoRunId, ProtoWorkflowId, decode_core_value, encode_core_value,
369 };
370 use crate::error::{ProtoWireError, WireError};
371
372 fn workflow_id() -> aion_core::WorkflowId {
373 aion_core::WorkflowId::new(uuid::Uuid::nil())
374 }
375
376 fn run_id() -> aion_core::RunId {
377 aion_core::RunId::new(uuid::Uuid::nil())
378 }
379
380 fn payload(label: &str) -> Result<ProtoPayload, aion_core::PayloadError> {
381 Ok(ProtoPayload::from(aion_core::Payload::from_json(
382 &json!({ "label": label }),
383 )?))
384 }
385
386 fn recorded_at() -> Result<DateTime<Utc>, chrono::ParseError> {
387 Ok(DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")?.with_timezone(&Utc))
388 }
389
390 fn assert_json_round_trip<T>(value: &T) -> Result<(), serde_json::Error>
391 where
392 T: Clone + PartialEq + serde::Serialize + DeserializeOwned,
393 {
394 let encoded = serde_json::to_string(value)?;
395 let decoded = serde_json::from_str::<T>(&encoded)?;
396 assert!(decoded == *value);
397 Ok(())
398 }
399
400 fn assert_proto_round_trip<T>(value: &T) -> Result<(), Box<dyn std::error::Error>>
401 where
402 T: Clone + PartialEq + Message + Default,
403 {
404 let mut bytes = Vec::new();
405 value.encode(&mut bytes)?;
406 let decoded = T::decode(bytes.as_slice())?;
407 assert!(decoded == *value);
408 Ok(())
409 }
410
411 #[test]
412 fn start_workflow_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
413 let request = ProtoStartWorkflowRequest {
414 namespace: String::from("tenant-a"),
415 workflow_type: String::from("checkout"),
416 input: Some(payload("input")?),
417 routing_key: Some(String::from("tenant-a/order-1")),
418 task_queue: Some(String::from("gpu")),
419 display_name: Some(String::from("Order 1 checkout")),
420 };
421 let response = ProtoStartWorkflowResponse {
422 workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
423 run_id: Some(ProtoRunId::from(run_id())),
424 };
425
426 assert_json_round_trip(&request)?;
427 assert_proto_round_trip(&request)?;
428 assert_json_round_trip(&response)?;
429 assert_proto_round_trip(&response)?;
430 Ok(())
431 }
432
433 #[test]
434 fn list_workflows_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
435 let list_request = WorkflowListRequest {
436 namespace: String::from("tenant-a"),
437 filter: WorkflowListFilter {
438 workflow_types: vec![String::from("checkout")],
439 statuses: vec![aion_core::WorkflowStatus::Running],
440 ..WorkflowListFilter::default()
441 },
442 sort: WorkflowSort {
443 field: WorkflowSortField::UpdatedAt,
444 direction: SortDirection::Desc,
445 },
446 cursor: Some(String::from("opaque")),
447 limit: 10,
448 };
449 let page = WorkflowListPage {
450 items: vec![WorkflowSummary {
451 workflow_id: workflow_id(),
452 run_id: run_id(),
453 workflow_type: String::from("checkout"),
454 status: aion_core::WorkflowStatus::Running,
455 started_at: recorded_at()?,
456 updated_at: recorded_at()?,
457 ended_at: None,
458 parent: None,
459 failed_step: None,
460 failure_reason: None,
461 display_name: Some(String::from("Nightly close")),
462 kind: None,
463 current_worker: None,
464 package_version: None,
465 }],
466 next_cursor: Some(String::from("next")),
467 count: 7,
468 provenance: None,
469 };
470 let request_envelope =
471 encode_core_value("tenant-a", Some(String::from("r1")), &list_request)?;
472 let page_envelope = encode_core_value("tenant-a", None, &page)?;
473 let request = ProtoListWorkflowsRequest {
474 namespace: String::from("tenant-a"),
475 request: Some(request_envelope.clone()),
476 };
477 let response = ProtoListWorkflowsResponse {
478 page: Some(page_envelope.clone()),
479 };
480
481 assert_json_round_trip(&request)?;
482 assert_proto_round_trip(&request)?;
483 assert_json_round_trip(&response)?;
484 assert_proto_round_trip(&response)?;
485 assert_eq!(
486 decode_core_value::<WorkflowListRequest>(&request_envelope)?,
487 list_request
488 );
489 assert_eq!(decode_core_value::<WorkflowListPage>(&page_envelope)?, page);
490 Ok(())
491 }
492
493 #[test]
494 fn query_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
495 let request = ProtoQueryRequest {
496 namespace: String::from("tenant-a"),
497 workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
498 run_id: Some(ProtoRunId::from(run_id())),
499 query_name: String::from("state"),
500 arguments: Some(payload("arguments")?),
501 };
502 let no_arguments_request = ProtoQueryRequest {
506 arguments: None,
507 ..request.clone()
508 };
509 let result_response = ProtoQueryResponse {
510 outcome: Some(proto_query_response::Outcome::Result(payload("result")?)),
511 };
512 let error_response = ProtoQueryResponse {
513 outcome: Some(proto_query_response::Outcome::Error(ProtoWireError::from(
514 WireError::unknown_query("state query is not registered"),
515 ))),
516 };
517
518 assert_json_round_trip(&request)?;
519 assert_proto_round_trip(&request)?;
520 assert_json_round_trip(&no_arguments_request)?;
521 assert_proto_round_trip(&no_arguments_request)?;
522 assert_json_round_trip(&result_response)?;
523 assert_proto_round_trip(&result_response)?;
524 assert_json_round_trip(&error_response)?;
525 assert_proto_round_trip(&error_response)?;
526 assert_ne!(request, no_arguments_request);
529 Ok(())
530 }
531
532 #[test]
535 fn start_workflow_display_name_absent_round_trips() -> Result<(), Box<dyn std::error::Error>> {
536 let named = ProtoStartWorkflowRequest {
537 namespace: String::from("tenant-a"),
538 workflow_type: String::from("checkout"),
539 input: Some(payload("input")?),
540 routing_key: None,
541 task_queue: None,
542 display_name: Some(String::from("Order 1 checkout")),
543 };
544 let unnamed = ProtoStartWorkflowRequest {
545 display_name: None,
546 ..named.clone()
547 };
548
549 assert_json_round_trip(&named)?;
550 assert_proto_round_trip(&named)?;
551 assert_json_round_trip(&unnamed)?;
552 assert_proto_round_trip(&unnamed)?;
553 assert_ne!(named, unnamed);
556 Ok(())
557 }
558
559 #[cfg(feature = "generated")]
569 #[test]
570 fn start_workflow_display_name_is_the_same_wire_field_as_the_generated_stub()
571 -> Result<(), Box<dyn std::error::Error>> {
572 const NAME: &str = "Nightly settlement";
573 let hand_written = ProtoStartWorkflowRequest {
574 namespace: String::from("tenant-a"),
575 workflow_type: String::from("checkout"),
576 input: None,
577 routing_key: None,
578 task_queue: None,
579 display_name: Some(String::from(NAME)),
580 };
581
582 let mut bytes = Vec::new();
584 hand_written.encode(&mut bytes)?;
585 let decoded = crate::generated::StartWorkflowRequest::decode(bytes.as_slice())?;
586 assert_eq!(
587 decoded.display_name.as_deref(),
588 Some(NAME),
589 "the generated stub must read the hand-written display_name"
590 );
591
592 let mut bytes = Vec::new();
594 decoded.encode(&mut bytes)?;
595 let round_tripped = ProtoStartWorkflowRequest::decode(bytes.as_slice())?;
596 assert_eq!(round_tripped, hand_written);
597
598 let mut bytes = Vec::new();
601 ProtoStartWorkflowRequest {
602 namespace: String::new(),
603 workflow_type: String::new(),
604 input: None,
605 routing_key: None,
606 task_queue: None,
607 display_name: Some(String::from("x")),
608 }
609 .encode(&mut bytes)?;
610 assert_eq!(bytes, vec![0x32, 0x01, b'x']);
611
612 let mut bytes = Vec::new();
613 ProtoStartWorkflowRequest {
614 namespace: String::new(),
615 workflow_type: String::new(),
616 input: None,
617 routing_key: None,
618 task_queue: None,
619 display_name: None,
620 }
621 .encode(&mut bytes)?;
622 assert!(
623 bytes.is_empty(),
624 "an unnamed start must put nothing on the wire, got {bytes:?}"
625 );
626 Ok(())
627 }
628
629 #[cfg(feature = "generated")]
632 #[test]
633 fn rename_messages_agree_with_the_generated_stubs() -> Result<(), Box<dyn std::error::Error>> {
634 let request = super::ProtoRenameRequest {
635 namespace: String::from("tenant-a"),
636 workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
637 run_id: Some(ProtoRunId::from(run_id())),
638 display_name: String::from("Nightly settlement"),
639 };
640 let mut bytes = Vec::new();
641 request.encode(&mut bytes)?;
642 let decoded = crate::generated::RenameRequest::decode(bytes.as_slice())?;
643 assert_eq!(decoded.display_name, "Nightly settlement");
644 assert_eq!(decoded.namespace, "tenant-a");
645 let mut bytes = Vec::new();
646 decoded.encode(&mut bytes)?;
647 assert_eq!(
648 super::ProtoRenameRequest::decode(bytes.as_slice())?,
649 request
650 );
651
652 let response = super::ProtoRenameResponse {
653 run_id: Some(ProtoRunId::from(run_id())),
654 display_name: String::from("Nightly settlement"),
655 };
656 let mut bytes = Vec::new();
657 response.encode(&mut bytes)?;
658 let decoded = crate::generated::RenameResponse::decode(bytes.as_slice())?;
659 assert_eq!(decoded.display_name, "Nightly settlement");
660 let mut bytes = Vec::new();
661 decoded.encode(&mut bytes)?;
662 assert_eq!(
663 super::ProtoRenameResponse::decode(bytes.as_slice())?,
664 response
665 );
666 Ok(())
667 }
668
669 #[test]
670 fn rename_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
671 let request = super::ProtoRenameRequest {
672 namespace: String::from("tenant-a"),
673 workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
674 run_id: Some(ProtoRunId::from(run_id())),
675 display_name: String::from("Nightly settlement"),
676 };
677 let response = super::ProtoRenameResponse {
678 run_id: Some(ProtoRunId::from(run_id())),
679 display_name: String::from("Nightly settlement"),
680 };
681
682 assert_json_round_trip(&request)?;
683 assert_proto_round_trip(&request)?;
684 assert_json_round_trip(&response)?;
685 assert_proto_round_trip(&response)?;
686 Ok(())
687 }
688
689 #[test]
690 fn reopen_round_trips_json_and_proto() -> Result<(), Box<dyn std::error::Error>> {
691 let request = ProtoReopenRequest {
692 namespace: String::from("tenant-a"),
693 workflow_id: Some(ProtoWorkflowId::from(workflow_id())),
694 run_id: Some(ProtoRunId::from(run_id())),
695 };
696 let response = ProtoReopenResponse {
697 run_id: Some(ProtoRunId::from(run_id())),
698 status: crate::convert::ProtoWorkflowStatus::Running as i32,
699 };
700
701 assert_json_round_trip(&request)?;
702 assert_proto_round_trip(&request)?;
703 assert_json_round_trip(&response)?;
704 assert_proto_round_trip(&response)?;
705 Ok(())
706 }
707}