1use async_trait::async_trait;
2use serde::{Deserialize, Serialize};
3
4use crate::CamelError;
5use crate::declarative::LanguageExpressionDef;
6use crate::splitter::StreamSplitConfig;
7
8pub const CANONICAL_CONTRACT_NAME: &str = "canonical-v1";
9pub const CANONICAL_CONTRACT_VERSION: u32 = 2;
10pub const CANONICAL_CONTRACT_SUPPORTED_STEPS: &[&str] = &[
11 "to",
12 "log",
13 "wire_tap",
14 "script",
15 "filter",
16 "choice",
17 "split",
18 "aggregate",
19 "stop",
20 "delay",
21];
22pub const CANONICAL_CONTRACT_DECLARATIVE_ONLY_STEPS: &[&str] =
23 &["script", "filter", "choice", "split"];
24pub const CANONICAL_CONTRACT_EXCLUDED_DECLARATIVE_STEPS: &[&str] = &[
25 "set_header",
26 "set_property",
27 "set_body",
28 "multicast",
29 "convert_body_to",
30 "bean",
31 "marshal",
32 "unmarshal",
33];
34pub const CANONICAL_CONTRACT_RUST_ONLY_STEPS: &[&str] = &[
35 "processor",
36 "process",
37 "process_fn",
38 "map_body",
39 "set_body_fn",
40 "set_header_fn",
41];
42
43pub fn canonical_contract_supports_step(step: &str) -> bool {
44 CANONICAL_CONTRACT_SUPPORTED_STEPS.contains(&step)
45}
46
47pub fn canonical_contract_rejection_reason(step: &str) -> Option<&'static str> {
48 if CANONICAL_CONTRACT_EXCLUDED_DECLARATIVE_STEPS.contains(&step) {
49 return Some(
50 "declared out-of-scope for canonical v1; use declarative route compilation path outside CQRS canonical commands",
51 );
52 }
53
54 if CANONICAL_CONTRACT_RUST_ONLY_STEPS.contains(&step) {
55 return Some("rust-only programmable step; not representable in canonical v1 contract");
56 }
57
58 if canonical_contract_supports_step(step)
59 && CANONICAL_CONTRACT_DECLARATIVE_ONLY_STEPS.contains(&step)
60 {
61 return Some(
62 "supported only as declarative/serializable expression form; closure/processor variants are outside canonical v1",
63 );
64 }
65
66 None
67}
68
69#[derive(
70 Debug,
71 Clone,
72 PartialEq,
73 Eq,
74 serde::Serialize,
75 serde::Deserialize,
76 schemars::JsonSchema,
77 ts_rs::TS,
78)]
79#[serde(rename_all = "snake_case")]
80#[ts(rename_all = "snake_case")]
81pub struct CanonicalRouteSpec {
82 pub route_id: String,
92 pub from: String,
93 pub steps: Vec<CanonicalStepSpec>,
94 pub circuit_breaker: Option<CanonicalCircuitBreakerSpec>,
95 pub auto_startup: Option<bool>,
96 pub startup_order: Option<i32>,
97 pub concurrency: Option<CanonicalConcurrencySpec>,
98 pub version: u32,
99}
100
101#[derive(
102 Debug,
103 Clone,
104 PartialEq,
105 Eq,
106 serde::Serialize,
107 serde::Deserialize,
108 schemars::JsonSchema,
109 ts_rs::TS,
110)]
111#[serde(tag = "step", content = "config", rename_all = "snake_case")]
112#[ts(rename_all = "snake_case")]
113pub enum CanonicalStepSpec {
114 To {
115 uri: String,
116 },
117 Log {
118 message: String,
119 },
120 WireTap {
121 uri: String,
122 },
123 Script {
124 expression: LanguageExpressionDef,
125 },
126 Filter {
127 predicate: LanguageExpressionDef,
128 steps: Vec<CanonicalStepSpec>,
129 },
130 Choice {
131 whens: Vec<CanonicalWhenSpec>,
132 otherwise: Option<Vec<CanonicalStepSpec>>,
133 },
134 Split {
135 expression: CanonicalSplitExpressionSpec,
136 aggregation: CanonicalSplitAggregationSpec,
137 parallel: bool,
138 parallel_limit: Option<usize>,
139 stop_on_exception: bool,
140 steps: Vec<CanonicalStepSpec>,
141 },
142 Aggregate(CanonicalAggregateSpec),
143 Stop,
144 Delay {
145 #[ts(type = "number")]
146 delay_ms: u64,
147 dynamic_header: Option<String>,
148 },
149}
150
151#[derive(
152 Debug,
153 Clone,
154 PartialEq,
155 Eq,
156 serde::Serialize,
157 serde::Deserialize,
158 schemars::JsonSchema,
159 ts_rs::TS,
160)]
161#[serde(rename_all = "snake_case")]
162#[ts(rename_all = "snake_case")]
163pub struct CanonicalWhenSpec {
164 pub predicate: LanguageExpressionDef,
165 pub steps: Vec<CanonicalStepSpec>,
166}
167
168#[derive(
169 Debug,
170 Clone,
171 PartialEq,
172 Eq,
173 serde::Serialize,
174 serde::Deserialize,
175 schemars::JsonSchema,
176 ts_rs::TS,
177)]
178#[serde(rename_all = "snake_case")]
179#[ts(rename_all = "snake_case")]
180pub enum CanonicalSplitExpressionSpec {
181 BodyLines,
182 BodyJsonArray,
183 Language(LanguageExpressionDef),
184 Stream(StreamSplitConfig),
185}
186
187#[derive(
188 Debug,
189 Clone,
190 PartialEq,
191 Eq,
192 serde::Serialize,
193 serde::Deserialize,
194 schemars::JsonSchema,
195 ts_rs::TS,
196)]
197#[serde(rename_all = "snake_case")]
198#[ts(rename_all = "snake_case")]
199pub enum CanonicalSplitAggregationSpec {
200 LastWins,
201 CollectAll,
202 Original,
203}
204
205#[derive(
206 Debug,
207 Clone,
208 PartialEq,
209 Eq,
210 serde::Serialize,
211 serde::Deserialize,
212 schemars::JsonSchema,
213 ts_rs::TS,
214)]
215#[serde(rename_all = "snake_case")]
216#[ts(rename_all = "snake_case")]
217pub enum CanonicalAggregateStrategySpec {
218 CollectAll,
219}
220
221#[derive(
222 Debug,
223 Clone,
224 PartialEq,
225 Eq,
226 serde::Serialize,
227 serde::Deserialize,
228 schemars::JsonSchema,
229 ts_rs::TS,
230)]
231#[serde(rename_all = "snake_case")]
232#[ts(rename_all = "snake_case")]
233pub struct CanonicalAggregateSpec {
234 pub header: String,
235 pub completion_size: Option<usize>,
236 #[ts(type = "number")]
237 pub completion_timeout_ms: Option<u64>,
238 pub correlation_key: Option<String>,
239 pub force_completion_on_stop: Option<bool>,
240 pub discard_on_timeout: Option<bool>,
241 pub strategy: CanonicalAggregateStrategySpec,
242 pub max_buckets: Option<usize>,
243 #[ts(type = "number")]
244 pub bucket_ttl_ms: Option<u64>,
245 #[serde(default)]
249 pub completion_predicate: Option<LanguageExpressionDef>,
250}
251
252#[derive(
253 Debug,
254 Clone,
255 PartialEq,
256 Eq,
257 serde::Serialize,
258 serde::Deserialize,
259 schemars::JsonSchema,
260 ts_rs::TS,
261)]
262#[serde(rename_all = "snake_case")]
263#[ts(rename_all = "snake_case")]
264pub struct CanonicalCircuitBreakerSpec {
265 pub failure_threshold: u32,
266 #[ts(type = "number")]
267 pub open_duration_ms: u64,
268}
269
270#[derive(
271 Debug,
272 Clone,
273 PartialEq,
274 Eq,
275 serde::Serialize,
276 serde::Deserialize,
277 schemars::JsonSchema,
278 ts_rs::TS,
279)]
280#[serde(tag = "mode", rename_all = "snake_case")]
281pub enum CanonicalConcurrencySpec {
282 Sequential,
283 Concurrent { max: usize },
284}
285
286impl CanonicalRouteSpec {
287 pub fn new(route_id: impl Into<String>, from: impl Into<String>) -> Self {
288 Self {
289 route_id: route_id.into(),
290 from: from.into(),
291 steps: Vec::new(),
292 circuit_breaker: None,
293 auto_startup: None,
294 startup_order: None,
295 concurrency: None,
296 version: CANONICAL_CONTRACT_VERSION,
297 }
298 }
299
300 pub fn with_auto_startup(mut self, auto: bool) -> Self {
301 self.auto_startup = Some(auto);
302 self
303 }
304
305 pub fn with_startup_order(mut self, order: i32) -> Self {
306 self.startup_order = Some(order);
307 self
308 }
309
310 pub fn with_concurrency(mut self, concurrency: CanonicalConcurrencySpec) -> Self {
311 self.concurrency = Some(concurrency);
312 self
313 }
314
315 pub fn validate_contract(&self) -> Result<(), CamelError> {
316 if self.route_id.trim().is_empty() {
317 return Err(CamelError::RouteError(
318 "canonical contract violation: route_id cannot be empty".to_string(),
319 ));
320 }
321 if self.from.trim().is_empty() {
322 return Err(CamelError::RouteError(
323 "canonical contract violation: from cannot be empty".to_string(),
324 ));
325 }
326 if self.version == 0 || self.version > CANONICAL_CONTRACT_VERSION {
327 return Err(CamelError::RouteError(format!(
328 "canonical contract violation: expected version {}, got {}",
329 CANONICAL_CONTRACT_VERSION, self.version
330 )));
331 }
332 validate_steps(&self.steps)?;
333 if let Some(cb) = &self.circuit_breaker {
334 if cb.failure_threshold == 0 {
335 return Err(CamelError::RouteError(
336 "canonical contract violation: circuit_breaker.failure_threshold must be > 0"
337 .to_string(),
338 ));
339 }
340 if cb.open_duration_ms == 0 {
341 return Err(CamelError::RouteError(
342 "canonical contract violation: circuit_breaker.open_duration_ms must be > 0"
343 .to_string(),
344 ));
345 }
346 }
347 if let Some(CanonicalConcurrencySpec::Concurrent { max: 0 }) = &self.concurrency {
348 return Err(CamelError::RouteError(
349 "canonical contract violation: concurrency max must be > 0".to_string(),
350 ));
351 }
352 Ok(())
353 }
354}
355
356#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
357pub struct CanonicalFieldLoss {
358 pub field: &'static str,
359 pub reason: String,
360 pub target_version: u32,
361}
362
363#[derive(Debug, Clone, PartialEq, Eq, Default, serde::Serialize)]
364pub struct CanonicalLossReport {
365 pub dropped_fields: Vec<CanonicalFieldLoss>,
366}
367
368impl CanonicalLossReport {
369 pub fn from_field(field: &'static str, reason: &str, target_version: u32) -> Self {
370 Self {
371 dropped_fields: vec![CanonicalFieldLoss {
372 field,
373 reason: reason.to_string(),
374 target_version,
375 }],
376 }
377 }
378
379 pub fn is_empty(&self) -> bool {
380 self.dropped_fields.is_empty()
381 }
382}
383
384fn validate_steps(steps: &[CanonicalStepSpec]) -> Result<(), CamelError> {
385 for step in steps {
386 match step {
387 CanonicalStepSpec::To { uri } | CanonicalStepSpec::WireTap { uri } => {
388 if uri.trim().is_empty() {
389 return Err(CamelError::RouteError(
390 "canonical contract violation: endpoint uri cannot be empty".to_string(),
391 ));
392 }
393 }
394 CanonicalStepSpec::Filter { steps, .. } => validate_steps(steps)?,
395 CanonicalStepSpec::Choice { whens, otherwise } => {
396 for when in whens {
397 validate_steps(&when.steps)?;
398 }
399 if let Some(otherwise) = otherwise {
400 validate_steps(otherwise)?;
401 }
402 }
403 CanonicalStepSpec::Split {
404 parallel_limit,
405 steps,
406 ..
407 } => {
408 if let Some(limit) = parallel_limit
409 && *limit == 0
410 {
411 return Err(CamelError::RouteError(
412 "canonical contract violation: split.parallel_limit must be > 0"
413 .to_string(),
414 ));
415 }
416 validate_steps(steps)?;
417 }
418 CanonicalStepSpec::Aggregate(config) => {
419 if config.header.trim().is_empty() {
420 return Err(CamelError::RouteError(
421 "canonical contract violation: aggregate.header cannot be empty"
422 .to_string(),
423 ));
424 }
425 if let Some(size) = config.completion_size
426 && size == 0
427 {
428 return Err(CamelError::RouteError(
429 "canonical contract violation: aggregate.completion_size must be > 0"
430 .to_string(),
431 ));
432 }
433 }
434 CanonicalStepSpec::Log { .. }
435 | CanonicalStepSpec::Script { .. }
436 | CanonicalStepSpec::Stop
437 | CanonicalStepSpec::Delay { .. } => {}
438 }
439 }
440 Ok(())
441}
442
443#[derive(Debug, Clone, PartialEq, Eq)]
444pub enum RuntimeCommand {
445 RegisterRoute {
446 spec: CanonicalRouteSpec,
447 command_id: String,
448 causation_id: Option<String>,
449 },
450 StartRoute {
451 route_id: String,
452 command_id: String,
453 causation_id: Option<String>,
454 },
455 StopRoute {
456 route_id: String,
457 command_id: String,
458 causation_id: Option<String>,
459 },
460 SuspendRoute {
461 route_id: String,
462 command_id: String,
463 causation_id: Option<String>,
464 },
465 ResumeRoute {
466 route_id: String,
467 command_id: String,
468 causation_id: Option<String>,
469 },
470 ReloadRoute {
471 route_id: String,
472 command_id: String,
473 causation_id: Option<String>,
474 },
475 FailRoute {
479 route_id: String,
480 error: String,
481 command_id: String,
482 causation_id: Option<String>,
483 },
484 RemoveRoute {
485 route_id: String,
486 command_id: String,
487 causation_id: Option<String>,
488 },
489 ReloadTlsCerts {
490 scheme: String,
491 host: String,
492 port: u16,
493 command_id: String,
494 causation_id: Option<String>,
495 },
496 ReloadTemplates {
503 route_id: String,
504 command_id: String,
505 causation_id: Option<String>,
506 },
507}
508
509impl RuntimeCommand {
510 pub fn command_id(&self) -> &str {
511 match self {
512 RuntimeCommand::RegisterRoute { command_id, .. }
513 | RuntimeCommand::StartRoute { command_id, .. }
514 | RuntimeCommand::StopRoute { command_id, .. }
515 | RuntimeCommand::SuspendRoute { command_id, .. }
516 | RuntimeCommand::ResumeRoute { command_id, .. }
517 | RuntimeCommand::ReloadRoute { command_id, .. }
518 | RuntimeCommand::FailRoute { command_id, .. }
519 | RuntimeCommand::RemoveRoute { command_id, .. }
520 | RuntimeCommand::ReloadTlsCerts { command_id, .. }
521 | RuntimeCommand::ReloadTemplates { command_id, .. } => command_id,
522 }
523 }
524
525 pub fn causation_id(&self) -> Option<&str> {
526 match self {
527 RuntimeCommand::RegisterRoute { causation_id, .. }
528 | RuntimeCommand::StartRoute { causation_id, .. }
529 | RuntimeCommand::StopRoute { causation_id, .. }
530 | RuntimeCommand::SuspendRoute { causation_id, .. }
531 | RuntimeCommand::ResumeRoute { causation_id, .. }
532 | RuntimeCommand::ReloadRoute { causation_id, .. }
533 | RuntimeCommand::FailRoute { causation_id, .. }
534 | RuntimeCommand::RemoveRoute { causation_id, .. }
535 | RuntimeCommand::ReloadTlsCerts { causation_id, .. }
536 | RuntimeCommand::ReloadTemplates { causation_id, .. } => causation_id.as_deref(),
537 }
538 }
539}
540
541#[derive(Debug, Clone, PartialEq, Eq)]
542pub enum RuntimeCommandResult {
543 Accepted,
544 Duplicate {
545 command_id: String,
546 },
547 RouteRegistered {
548 route_id: String,
549 },
550 RouteStateChanged {
551 route_id: String,
552 status: String,
553 },
554 TlsCertsReloaded {
555 scheme: String,
556 host: String,
557 port: u16,
558 },
559 TemplatesReloaded {
560 route_id: String,
561 },
562}
563
564#[derive(Debug, Clone, PartialEq, Eq)]
565pub enum RuntimeQuery {
566 GetRouteStatus {
567 route_id: String,
568 },
569 InFlightCount {
573 route_id: String,
574 },
575 ListRoutes,
576}
577
578#[derive(Debug, Clone, PartialEq, Eq)]
579pub enum RuntimeQueryResult {
580 InFlightCount { route_id: String, count: u64 },
581 RouteNotFound { route_id: String },
582 RouteStatus { route_id: String, status: String },
583 Routes { route_ids: Vec<String> },
584}
585
586#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
587pub enum RuntimeEvent {
588 RouteRegistered { route_id: String },
589 RouteStartRequested { route_id: String },
590 RouteStarted { route_id: String },
591 RouteFailed { route_id: String, error: String },
592 RouteStopped { route_id: String },
593 RouteSuspended { route_id: String },
594 RouteResumed { route_id: String },
595 RouteReloaded { route_id: String },
596 RouteRemoved { route_id: String },
597}
598
599#[async_trait]
600pub trait RuntimeCommandBus: Send + Sync {
601 async fn execute(&self, cmd: RuntimeCommand) -> Result<RuntimeCommandResult, CamelError>;
602}
603
604#[async_trait]
605pub trait RuntimeQueryBus: Send + Sync {
606 async fn ask(&self, query: RuntimeQuery) -> Result<RuntimeQueryResult, CamelError>;
607}
608
609pub trait RuntimeHandle: RuntimeCommandBus + RuntimeQueryBus {}
610
611impl<T> RuntimeHandle for T where T: RuntimeCommandBus + RuntimeQueryBus {}
612
613#[cfg(test)]
614mod tests {
615 use super::*;
616 use async_trait::async_trait;
617 use futures::executor::block_on;
618
619 struct NoopRuntime;
620
621 #[async_trait]
622 impl RuntimeCommandBus for NoopRuntime {
623 async fn execute(&self, cmd: RuntimeCommand) -> Result<RuntimeCommandResult, CamelError> {
624 Ok(match cmd {
625 RuntimeCommand::RegisterRoute { spec, .. } => {
626 RuntimeCommandResult::RouteRegistered {
627 route_id: spec.route_id,
628 }
629 }
630 RuntimeCommand::StartRoute { route_id, .. }
631 | RuntimeCommand::StopRoute { route_id, .. }
632 | RuntimeCommand::SuspendRoute { route_id, .. }
633 | RuntimeCommand::ResumeRoute { route_id, .. }
634 | RuntimeCommand::ReloadRoute { route_id, .. }
635 | RuntimeCommand::FailRoute { route_id, .. }
636 | RuntimeCommand::RemoveRoute { route_id, .. } => {
637 RuntimeCommandResult::RouteStateChanged {
638 route_id,
639 status: "ok".to_string(),
640 }
641 }
642 RuntimeCommand::ReloadTlsCerts {
643 scheme, host, port, ..
644 } => RuntimeCommandResult::TlsCertsReloaded { scheme, host, port },
645 RuntimeCommand::ReloadTemplates { route_id, .. } => {
646 RuntimeCommandResult::TemplatesReloaded { route_id }
647 }
648 })
649 }
650 }
651
652 #[async_trait]
653 impl RuntimeQueryBus for NoopRuntime {
654 async fn ask(&self, query: RuntimeQuery) -> Result<RuntimeQueryResult, CamelError> {
655 Ok(match query {
656 RuntimeQuery::GetRouteStatus { route_id } => RuntimeQueryResult::RouteStatus {
657 route_id,
658 status: "Started".to_string(),
659 },
660 RuntimeQuery::InFlightCount { route_id } => {
661 RuntimeQueryResult::InFlightCount { route_id, count: 0 }
662 }
663 RuntimeQuery::ListRoutes => RuntimeQueryResult::Routes {
664 route_ids: vec!["r1".to_string()],
665 },
666 })
667 }
668 }
669
670 #[test]
671 fn command_and_query_ids_are_exposed() {
672 let cmd = RuntimeCommand::StartRoute {
673 route_id: "r1".into(),
674 command_id: "c1".into(),
675 causation_id: None,
676 };
677 assert_eq!(cmd.command_id(), "c1");
678 }
679
680 #[test]
681 fn canonical_spec_requires_route_id_and_from() {
682 let spec = CanonicalRouteSpec::new("r1", "timer:tick");
683 assert_eq!(spec.route_id, "r1");
684 assert_eq!(spec.from, "timer:tick");
685 assert_eq!(spec.version, CANONICAL_CONTRACT_VERSION);
686 assert!(spec.steps.is_empty());
687 assert!(spec.circuit_breaker.is_none());
688 }
689
690 #[test]
691 fn canonical_contract_rejects_invalid_version() {
692 let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
693 spec.version = 3;
694 let err = spec.validate_contract().unwrap_err().to_string();
695 assert!(err.contains("expected version"));
696 }
697
698 #[test]
699 fn canonical_contract_declares_subset_scope() {
700 assert!(canonical_contract_supports_step("to"));
701 assert!(canonical_contract_supports_step("split"));
702 assert!(!canonical_contract_supports_step("set_header"));
703 assert!(!canonical_contract_supports_step("set_property"));
704
705 assert!(CANONICAL_CONTRACT_DECLARATIVE_ONLY_STEPS.contains(&"split"));
706 assert!(CANONICAL_CONTRACT_EXCLUDED_DECLARATIVE_STEPS.contains(&"set_header"));
707 assert!(CANONICAL_CONTRACT_EXCLUDED_DECLARATIVE_STEPS.contains(&"set_property"));
708 assert!(CANONICAL_CONTRACT_RUST_ONLY_STEPS.contains(&"processor"));
709 }
710
711 #[test]
712 fn canonical_contract_rejection_reason_is_explicit() {
713 let set_header_reason = canonical_contract_rejection_reason("set_header")
714 .expect("set_header should have explicit reason");
715 assert!(set_header_reason.contains("out-of-scope"));
716
717 let set_property_reason = canonical_contract_rejection_reason("set_property")
718 .expect("set_property should have explicit reason");
719 assert!(set_property_reason.contains("out-of-scope"));
720
721 let processor_reason = canonical_contract_rejection_reason("processor")
722 .expect("processor should be rust-only");
723 assert!(processor_reason.contains("rust-only"));
724
725 let split_reason = canonical_contract_rejection_reason("split")
726 .expect("split should require declarative form");
727 assert!(split_reason.contains("declarative"));
728 }
729
730 #[test]
731 fn command_causation_id_is_exposed() {
732 let cmd = RuntimeCommand::StopRoute {
733 route_id: "r1".into(),
734 command_id: "c2".into(),
735 causation_id: Some("c1".into()),
736 };
737 assert_eq!(cmd.command_id(), "c2");
738 assert_eq!(cmd.causation_id(), Some("c1"));
739 }
740
741 #[test]
742 fn canonical_contract_rejects_empty_route_id_and_from() {
743 let spec = CanonicalRouteSpec::new(" ", "timer:tick");
744 let err = spec.validate_contract().unwrap_err().to_string();
745 assert!(err.contains("route_id cannot be empty"));
746
747 let spec = CanonicalRouteSpec::new("r1", " ");
748 let err = spec.validate_contract().unwrap_err().to_string();
749 assert!(err.contains("from cannot be empty"));
750 }
751
752 #[test]
753 fn canonical_contract_rejects_invalid_nested_steps() {
754 let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
755 spec.steps = vec![CanonicalStepSpec::Split {
756 expression: CanonicalSplitExpressionSpec::BodyLines,
757 aggregation: CanonicalSplitAggregationSpec::CollectAll,
758 parallel: true,
759 parallel_limit: Some(0),
760 stop_on_exception: false,
761 steps: vec![CanonicalStepSpec::To {
762 uri: "log:ok".to_string(),
763 }],
764 }];
765 let err = spec.validate_contract().unwrap_err().to_string();
766 assert!(err.contains("split.parallel_limit must be > 0"));
767
768 spec.steps = vec![CanonicalStepSpec::To {
769 uri: " ".to_string(),
770 }];
771 let err = spec.validate_contract().unwrap_err().to_string();
772 assert!(err.contains("endpoint uri cannot be empty"));
773 }
774
775 #[test]
776 fn canonical_contract_rejects_invalid_aggregate_and_circuit_breaker() {
777 let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
778 spec.steps = vec![CanonicalStepSpec::Aggregate(CanonicalAggregateSpec {
779 header: " ".to_string(),
780 completion_size: Some(1),
781 completion_timeout_ms: None,
782 correlation_key: None,
783 force_completion_on_stop: None,
784 discard_on_timeout: None,
785 strategy: CanonicalAggregateStrategySpec::CollectAll,
786 max_buckets: None,
787 bucket_ttl_ms: None,
788 completion_predicate: None,
789 })];
790 let err = spec.validate_contract().unwrap_err().to_string();
791 assert!(err.contains("aggregate.header cannot be empty"));
792
793 spec.steps = vec![CanonicalStepSpec::Aggregate(CanonicalAggregateSpec {
794 header: "k".to_string(),
795 completion_size: Some(0),
796 completion_timeout_ms: None,
797 correlation_key: None,
798 force_completion_on_stop: None,
799 discard_on_timeout: None,
800 strategy: CanonicalAggregateStrategySpec::CollectAll,
801 max_buckets: None,
802 bucket_ttl_ms: None,
803 completion_predicate: None,
804 })];
805 let err = spec.validate_contract().unwrap_err().to_string();
806 assert!(err.contains("aggregate.completion_size must be > 0"));
807
808 spec.steps = vec![];
809 spec.circuit_breaker = Some(CanonicalCircuitBreakerSpec {
810 failure_threshold: 0,
811 open_duration_ms: 10,
812 });
813 let err = spec.validate_contract().unwrap_err().to_string();
814 assert!(err.contains("failure_threshold must be > 0"));
815
816 spec.circuit_breaker = Some(CanonicalCircuitBreakerSpec {
817 failure_threshold: 1,
818 open_duration_ms: 0,
819 });
820 let err = spec.validate_contract().unwrap_err().to_string();
821 assert!(err.contains("open_duration_ms must be > 0"));
822 }
823
824 #[test]
825 fn canonical_contract_rejection_reason_none_for_regular_steps() {
826 assert!(canonical_contract_rejection_reason("to").is_none());
827 assert!(canonical_contract_rejection_reason("unknown-step").is_none());
828 }
829
830 #[test]
831 fn command_helpers_cover_all_variants() {
832 let spec = CanonicalRouteSpec::new("r1", "timer:tick");
833 let cmds = [
834 RuntimeCommand::RegisterRoute {
835 spec,
836 command_id: "c1".into(),
837 causation_id: Some("root".into()),
838 },
839 RuntimeCommand::StartRoute {
840 route_id: "r1".into(),
841 command_id: "c2".into(),
842 causation_id: None,
843 },
844 RuntimeCommand::StopRoute {
845 route_id: "r1".into(),
846 command_id: "c3".into(),
847 causation_id: None,
848 },
849 RuntimeCommand::SuspendRoute {
850 route_id: "r1".into(),
851 command_id: "c4".into(),
852 causation_id: None,
853 },
854 RuntimeCommand::ResumeRoute {
855 route_id: "r1".into(),
856 command_id: "c5".into(),
857 causation_id: None,
858 },
859 RuntimeCommand::ReloadRoute {
860 route_id: "r1".into(),
861 command_id: "c6".into(),
862 causation_id: None,
863 },
864 RuntimeCommand::FailRoute {
865 route_id: "r1".into(),
866 error: "boom".into(),
867 command_id: "c7".into(),
868 causation_id: None,
869 },
870 RuntimeCommand::RemoveRoute {
871 route_id: "r1".into(),
872 command_id: "c8".into(),
873 causation_id: None,
874 },
875 RuntimeCommand::ReloadTlsCerts {
876 scheme: "https".into(),
877 host: "example.com".into(),
878 port: 8443,
879 command_id: "c9".into(),
880 causation_id: None,
881 },
882 ];
883
884 let ids: Vec<&str> = cmds.iter().map(RuntimeCommand::command_id).collect();
885 assert_eq!(
886 ids,
887 vec!["c1", "c2", "c3", "c4", "c5", "c6", "c7", "c8", "c9"]
888 );
889 assert_eq!(cmds[0].causation_id(), Some("root"));
890 assert_eq!(cmds[1].causation_id(), None);
891 }
892
893 #[test]
894 fn canonical_route_spec_serde_roundtrip() {
895 let mut spec = CanonicalRouteSpec::new("test-route", "timer:tick?period=1000");
896 spec.steps.push(CanonicalStepSpec::Log {
897 message: "Hello".into(),
898 });
899 spec.steps.push(CanonicalStepSpec::To {
900 uri: "log:info".into(),
901 });
902 spec.steps.push(CanonicalStepSpec::Stop);
903
904 let json = serde_json::to_string(&spec).unwrap();
905 let deserialized: CanonicalRouteSpec = serde_json::from_str(&json).unwrap();
906 assert_eq!(spec, deserialized);
907 }
908
909 #[test]
910 fn canonical_step_spec_serde_variants() {
911 let steps = vec![
912 CanonicalStepSpec::To {
913 uri: "direct:a".into(),
914 },
915 CanonicalStepSpec::Log {
916 message: "msg".into(),
917 },
918 CanonicalStepSpec::WireTap {
919 uri: "direct:audit".into(),
920 },
921 CanonicalStepSpec::Stop,
922 CanonicalStepSpec::Delay {
923 delay_ms: 100,
924 dynamic_header: None,
925 },
926 ];
927 let json = serde_json::to_string_pretty(&steps).unwrap();
928 let back: Vec<CanonicalStepSpec> = serde_json::from_str(&json).unwrap();
929 assert_eq!(steps, back);
930 }
931
932 #[test]
933 fn canonical_route_spec_json_schema_generates() {
934 let schema = schemars::schema_for!(CanonicalRouteSpec);
935 let json = serde_json::to_string(&schema).unwrap();
936 assert!(json.contains("CanonicalRouteSpec"));
937 assert!(json.contains("route_id"));
938 }
939
940 #[test]
941 fn canonical_json_schema_has_no_function_step() {
942 let schema = schemars::schema_for!(CanonicalRouteSpec);
943 let json = serde_json::to_string(&schema).unwrap();
944 assert!(
945 !json.contains("\"function\""),
946 "canonical JSON schema must not contain 'function' step"
947 );
948 }
949
950 #[test]
951 fn canonical_contract_does_not_support_function() {
952 assert!(
953 !canonical_contract_supports_step("function"),
954 "function must not be in CANONICAL_CONTRACT_SUPPORTED_STEPS"
955 );
956 }
957
958 #[test]
959 fn runtime_command_result_all_variants_are_distinct() {
960 let accepted = RuntimeCommandResult::Accepted;
961 let dup = RuntimeCommandResult::Duplicate {
962 command_id: "c1".into(),
963 };
964 let registered = RuntimeCommandResult::RouteRegistered {
965 route_id: "r1".into(),
966 };
967 let changed = RuntimeCommandResult::RouteStateChanged {
968 route_id: "r1".into(),
969 status: "Started".into(),
970 };
971
972 assert_ne!(accepted, dup);
973 assert_ne!(dup, registered);
974 assert_ne!(registered, changed);
975
976 let dup2 = RuntimeCommandResult::Duplicate {
977 command_id: "c1".into(),
978 };
979 assert_eq!(dup, dup2);
980 }
981
982 #[test]
983 fn runtime_event_serialization_round_trip() {
984 let event = RuntimeEvent::RouteFailed {
985 route_id: "route-a".to_string(),
986 error: "boom".to_string(),
987 };
988 let json = serde_json::to_string(&event).unwrap();
989 let back: RuntimeEvent = serde_json::from_str(&json).unwrap();
990 assert_eq!(event, back);
991 }
992
993 #[test]
994 fn noop_runtime_execute_and_ask_return_expected_shapes() {
995 let rt = NoopRuntime;
996 let cmd = RuntimeCommand::RegisterRoute {
997 spec: CanonicalRouteSpec::new("r2", "timer:tick"),
998 command_id: "c1".into(),
999 causation_id: None,
1000 };
1001 let cmd_result = block_on(rt.execute(cmd)).unwrap();
1002 assert_eq!(
1003 cmd_result,
1004 RuntimeCommandResult::RouteRegistered {
1005 route_id: "r2".into()
1006 }
1007 );
1008
1009 let query_result = block_on(rt.ask(RuntimeQuery::GetRouteStatus {
1010 route_id: "r2".into(),
1011 }))
1012 .unwrap();
1013 assert_eq!(
1014 query_result,
1015 RuntimeQueryResult::RouteStatus {
1016 route_id: "r2".into(),
1017 status: "Started".into()
1018 }
1019 );
1020 }
1021
1022 #[test]
1023 fn canonical_contract_name_and_version_constants_match() {
1024 assert_eq!(CANONICAL_CONTRACT_NAME, "canonical-v1");
1025 assert_eq!(CANONICAL_CONTRACT_VERSION, 2);
1026 }
1027
1028 #[test]
1029 fn canonical_concurrency_spec_rejects_zero_max() {
1030 let spec = CanonicalRouteSpec::new("r1", "timer:tick")
1031 .with_concurrency(CanonicalConcurrencySpec::Concurrent { max: 0 });
1032 let err = spec.validate_contract().unwrap_err().to_string();
1033 assert!(err.contains("concurrency max must be > 0"), "{err}");
1034 }
1035
1036 #[test]
1037 fn canonical_v2_round_trip() {
1038 let spec = CanonicalRouteSpec::new("r1", "timer:tick")
1039 .with_auto_startup(false)
1040 .with_startup_order(42)
1041 .with_concurrency(CanonicalConcurrencySpec::Concurrent { max: 8 });
1042 spec.validate_contract().unwrap();
1043 }
1044
1045 #[test]
1046 fn canonical_v2_version_is_2() {
1047 assert_eq!(CANONICAL_CONTRACT_VERSION, 2);
1048 }
1049
1050 #[test]
1051 fn canonical_loss_report_builder() {
1052 let report =
1053 CanonicalLossReport::from_field("error_handler", "not supported by canonical path", 2);
1054 assert_eq!(report.dropped_fields.len(), 1);
1055 assert_eq!(report.dropped_fields[0].field, "error_handler");
1056 }
1057
1058 #[test]
1059 fn canonical_v1_json_deserializes_in_v2() {
1060 let json = r#"{"route_id":"r1","from":"timer:tick","steps":[],"version":1}"#;
1061 let spec: CanonicalRouteSpec = serde_json::from_str(json).unwrap();
1062 assert_eq!(spec.route_id, "r1");
1063 assert!(spec.auto_startup.is_none());
1064 assert!(spec.startup_order.is_none());
1065 assert!(spec.concurrency.is_none());
1066 spec.validate_contract().unwrap();
1068 }
1069}