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}
497
498impl RuntimeCommand {
499 pub fn command_id(&self) -> &str {
500 match self {
501 RuntimeCommand::RegisterRoute { command_id, .. }
502 | RuntimeCommand::StartRoute { command_id, .. }
503 | RuntimeCommand::StopRoute { command_id, .. }
504 | RuntimeCommand::SuspendRoute { command_id, .. }
505 | RuntimeCommand::ResumeRoute { command_id, .. }
506 | RuntimeCommand::ReloadRoute { command_id, .. }
507 | RuntimeCommand::FailRoute { command_id, .. }
508 | RuntimeCommand::RemoveRoute { command_id, .. }
509 | RuntimeCommand::ReloadTlsCerts { command_id, .. } => command_id,
510 }
511 }
512
513 pub fn causation_id(&self) -> Option<&str> {
514 match self {
515 RuntimeCommand::RegisterRoute { causation_id, .. }
516 | RuntimeCommand::StartRoute { causation_id, .. }
517 | RuntimeCommand::StopRoute { causation_id, .. }
518 | RuntimeCommand::SuspendRoute { causation_id, .. }
519 | RuntimeCommand::ResumeRoute { causation_id, .. }
520 | RuntimeCommand::ReloadRoute { causation_id, .. }
521 | RuntimeCommand::FailRoute { causation_id, .. }
522 | RuntimeCommand::RemoveRoute { causation_id, .. }
523 | RuntimeCommand::ReloadTlsCerts { causation_id, .. } => causation_id.as_deref(),
524 }
525 }
526}
527
528#[derive(Debug, Clone, PartialEq, Eq)]
529pub enum RuntimeCommandResult {
530 Accepted,
531 Duplicate {
532 command_id: String,
533 },
534 RouteRegistered {
535 route_id: String,
536 },
537 RouteStateChanged {
538 route_id: String,
539 status: String,
540 },
541 TlsCertsReloaded {
542 scheme: String,
543 host: String,
544 port: u16,
545 },
546}
547
548#[derive(Debug, Clone, PartialEq, Eq)]
549pub enum RuntimeQuery {
550 GetRouteStatus {
551 route_id: String,
552 },
553 InFlightCount {
557 route_id: String,
558 },
559 ListRoutes,
560}
561
562#[derive(Debug, Clone, PartialEq, Eq)]
563pub enum RuntimeQueryResult {
564 InFlightCount { route_id: String, count: u64 },
565 RouteNotFound { route_id: String },
566 RouteStatus { route_id: String, status: String },
567 Routes { route_ids: Vec<String> },
568}
569
570#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
571pub enum RuntimeEvent {
572 RouteRegistered { route_id: String },
573 RouteStartRequested { route_id: String },
574 RouteStarted { route_id: String },
575 RouteFailed { route_id: String, error: String },
576 RouteStopped { route_id: String },
577 RouteSuspended { route_id: String },
578 RouteResumed { route_id: String },
579 RouteReloaded { route_id: String },
580 RouteRemoved { route_id: String },
581}
582
583#[async_trait]
584pub trait RuntimeCommandBus: Send + Sync {
585 async fn execute(&self, cmd: RuntimeCommand) -> Result<RuntimeCommandResult, CamelError>;
586}
587
588#[async_trait]
589pub trait RuntimeQueryBus: Send + Sync {
590 async fn ask(&self, query: RuntimeQuery) -> Result<RuntimeQueryResult, CamelError>;
591}
592
593pub trait RuntimeHandle: RuntimeCommandBus + RuntimeQueryBus {}
594
595impl<T> RuntimeHandle for T where T: RuntimeCommandBus + RuntimeQueryBus {}
596
597#[cfg(test)]
598mod tests {
599 use super::*;
600 use async_trait::async_trait;
601 use futures::executor::block_on;
602
603 struct NoopRuntime;
604
605 #[async_trait]
606 impl RuntimeCommandBus for NoopRuntime {
607 async fn execute(&self, cmd: RuntimeCommand) -> Result<RuntimeCommandResult, CamelError> {
608 Ok(match cmd {
609 RuntimeCommand::RegisterRoute { spec, .. } => {
610 RuntimeCommandResult::RouteRegistered {
611 route_id: spec.route_id,
612 }
613 }
614 RuntimeCommand::StartRoute { route_id, .. }
615 | RuntimeCommand::StopRoute { route_id, .. }
616 | RuntimeCommand::SuspendRoute { route_id, .. }
617 | RuntimeCommand::ResumeRoute { route_id, .. }
618 | RuntimeCommand::ReloadRoute { route_id, .. }
619 | RuntimeCommand::FailRoute { route_id, .. }
620 | RuntimeCommand::RemoveRoute { route_id, .. } => {
621 RuntimeCommandResult::RouteStateChanged {
622 route_id,
623 status: "ok".to_string(),
624 }
625 }
626 RuntimeCommand::ReloadTlsCerts {
627 scheme, host, port, ..
628 } => RuntimeCommandResult::TlsCertsReloaded { scheme, host, port },
629 })
630 }
631 }
632
633 #[async_trait]
634 impl RuntimeQueryBus for NoopRuntime {
635 async fn ask(&self, query: RuntimeQuery) -> Result<RuntimeQueryResult, CamelError> {
636 Ok(match query {
637 RuntimeQuery::GetRouteStatus { route_id } => RuntimeQueryResult::RouteStatus {
638 route_id,
639 status: "Started".to_string(),
640 },
641 RuntimeQuery::InFlightCount { route_id } => {
642 RuntimeQueryResult::InFlightCount { route_id, count: 0 }
643 }
644 RuntimeQuery::ListRoutes => RuntimeQueryResult::Routes {
645 route_ids: vec!["r1".to_string()],
646 },
647 })
648 }
649 }
650
651 #[test]
652 fn command_and_query_ids_are_exposed() {
653 let cmd = RuntimeCommand::StartRoute {
654 route_id: "r1".into(),
655 command_id: "c1".into(),
656 causation_id: None,
657 };
658 assert_eq!(cmd.command_id(), "c1");
659 }
660
661 #[test]
662 fn canonical_spec_requires_route_id_and_from() {
663 let spec = CanonicalRouteSpec::new("r1", "timer:tick");
664 assert_eq!(spec.route_id, "r1");
665 assert_eq!(spec.from, "timer:tick");
666 assert_eq!(spec.version, CANONICAL_CONTRACT_VERSION);
667 assert!(spec.steps.is_empty());
668 assert!(spec.circuit_breaker.is_none());
669 }
670
671 #[test]
672 fn canonical_contract_rejects_invalid_version() {
673 let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
674 spec.version = 3;
675 let err = spec.validate_contract().unwrap_err().to_string();
676 assert!(err.contains("expected version"));
677 }
678
679 #[test]
680 fn canonical_contract_declares_subset_scope() {
681 assert!(canonical_contract_supports_step("to"));
682 assert!(canonical_contract_supports_step("split"));
683 assert!(!canonical_contract_supports_step("set_header"));
684 assert!(!canonical_contract_supports_step("set_property"));
685
686 assert!(CANONICAL_CONTRACT_DECLARATIVE_ONLY_STEPS.contains(&"split"));
687 assert!(CANONICAL_CONTRACT_EXCLUDED_DECLARATIVE_STEPS.contains(&"set_header"));
688 assert!(CANONICAL_CONTRACT_EXCLUDED_DECLARATIVE_STEPS.contains(&"set_property"));
689 assert!(CANONICAL_CONTRACT_RUST_ONLY_STEPS.contains(&"processor"));
690 }
691
692 #[test]
693 fn canonical_contract_rejection_reason_is_explicit() {
694 let set_header_reason = canonical_contract_rejection_reason("set_header")
695 .expect("set_header should have explicit reason");
696 assert!(set_header_reason.contains("out-of-scope"));
697
698 let set_property_reason = canonical_contract_rejection_reason("set_property")
699 .expect("set_property should have explicit reason");
700 assert!(set_property_reason.contains("out-of-scope"));
701
702 let processor_reason = canonical_contract_rejection_reason("processor")
703 .expect("processor should be rust-only");
704 assert!(processor_reason.contains("rust-only"));
705
706 let split_reason = canonical_contract_rejection_reason("split")
707 .expect("split should require declarative form");
708 assert!(split_reason.contains("declarative"));
709 }
710
711 #[test]
712 fn command_causation_id_is_exposed() {
713 let cmd = RuntimeCommand::StopRoute {
714 route_id: "r1".into(),
715 command_id: "c2".into(),
716 causation_id: Some("c1".into()),
717 };
718 assert_eq!(cmd.command_id(), "c2");
719 assert_eq!(cmd.causation_id(), Some("c1"));
720 }
721
722 #[test]
723 fn canonical_contract_rejects_empty_route_id_and_from() {
724 let spec = CanonicalRouteSpec::new(" ", "timer:tick");
725 let err = spec.validate_contract().unwrap_err().to_string();
726 assert!(err.contains("route_id cannot be empty"));
727
728 let spec = CanonicalRouteSpec::new("r1", " ");
729 let err = spec.validate_contract().unwrap_err().to_string();
730 assert!(err.contains("from cannot be empty"));
731 }
732
733 #[test]
734 fn canonical_contract_rejects_invalid_nested_steps() {
735 let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
736 spec.steps = vec![CanonicalStepSpec::Split {
737 expression: CanonicalSplitExpressionSpec::BodyLines,
738 aggregation: CanonicalSplitAggregationSpec::CollectAll,
739 parallel: true,
740 parallel_limit: Some(0),
741 stop_on_exception: false,
742 steps: vec![CanonicalStepSpec::To {
743 uri: "log:ok".to_string(),
744 }],
745 }];
746 let err = spec.validate_contract().unwrap_err().to_string();
747 assert!(err.contains("split.parallel_limit must be > 0"));
748
749 spec.steps = vec![CanonicalStepSpec::To {
750 uri: " ".to_string(),
751 }];
752 let err = spec.validate_contract().unwrap_err().to_string();
753 assert!(err.contains("endpoint uri cannot be empty"));
754 }
755
756 #[test]
757 fn canonical_contract_rejects_invalid_aggregate_and_circuit_breaker() {
758 let mut spec = CanonicalRouteSpec::new("r1", "timer:tick");
759 spec.steps = vec![CanonicalStepSpec::Aggregate(CanonicalAggregateSpec {
760 header: " ".to_string(),
761 completion_size: Some(1),
762 completion_timeout_ms: None,
763 correlation_key: None,
764 force_completion_on_stop: None,
765 discard_on_timeout: None,
766 strategy: CanonicalAggregateStrategySpec::CollectAll,
767 max_buckets: None,
768 bucket_ttl_ms: None,
769 completion_predicate: None,
770 })];
771 let err = spec.validate_contract().unwrap_err().to_string();
772 assert!(err.contains("aggregate.header cannot be empty"));
773
774 spec.steps = vec![CanonicalStepSpec::Aggregate(CanonicalAggregateSpec {
775 header: "k".to_string(),
776 completion_size: Some(0),
777 completion_timeout_ms: None,
778 correlation_key: None,
779 force_completion_on_stop: None,
780 discard_on_timeout: None,
781 strategy: CanonicalAggregateStrategySpec::CollectAll,
782 max_buckets: None,
783 bucket_ttl_ms: None,
784 completion_predicate: None,
785 })];
786 let err = spec.validate_contract().unwrap_err().to_string();
787 assert!(err.contains("aggregate.completion_size must be > 0"));
788
789 spec.steps = vec![];
790 spec.circuit_breaker = Some(CanonicalCircuitBreakerSpec {
791 failure_threshold: 0,
792 open_duration_ms: 10,
793 });
794 let err = spec.validate_contract().unwrap_err().to_string();
795 assert!(err.contains("failure_threshold must be > 0"));
796
797 spec.circuit_breaker = Some(CanonicalCircuitBreakerSpec {
798 failure_threshold: 1,
799 open_duration_ms: 0,
800 });
801 let err = spec.validate_contract().unwrap_err().to_string();
802 assert!(err.contains("open_duration_ms must be > 0"));
803 }
804
805 #[test]
806 fn canonical_contract_rejection_reason_none_for_regular_steps() {
807 assert!(canonical_contract_rejection_reason("to").is_none());
808 assert!(canonical_contract_rejection_reason("unknown-step").is_none());
809 }
810
811 #[test]
812 fn command_helpers_cover_all_variants() {
813 let spec = CanonicalRouteSpec::new("r1", "timer:tick");
814 let cmds = [
815 RuntimeCommand::RegisterRoute {
816 spec,
817 command_id: "c1".into(),
818 causation_id: Some("root".into()),
819 },
820 RuntimeCommand::StartRoute {
821 route_id: "r1".into(),
822 command_id: "c2".into(),
823 causation_id: None,
824 },
825 RuntimeCommand::StopRoute {
826 route_id: "r1".into(),
827 command_id: "c3".into(),
828 causation_id: None,
829 },
830 RuntimeCommand::SuspendRoute {
831 route_id: "r1".into(),
832 command_id: "c4".into(),
833 causation_id: None,
834 },
835 RuntimeCommand::ResumeRoute {
836 route_id: "r1".into(),
837 command_id: "c5".into(),
838 causation_id: None,
839 },
840 RuntimeCommand::ReloadRoute {
841 route_id: "r1".into(),
842 command_id: "c6".into(),
843 causation_id: None,
844 },
845 RuntimeCommand::FailRoute {
846 route_id: "r1".into(),
847 error: "boom".into(),
848 command_id: "c7".into(),
849 causation_id: None,
850 },
851 RuntimeCommand::RemoveRoute {
852 route_id: "r1".into(),
853 command_id: "c8".into(),
854 causation_id: None,
855 },
856 RuntimeCommand::ReloadTlsCerts {
857 scheme: "https".into(),
858 host: "example.com".into(),
859 port: 8443,
860 command_id: "c9".into(),
861 causation_id: None,
862 },
863 ];
864
865 let ids: Vec<&str> = cmds.iter().map(RuntimeCommand::command_id).collect();
866 assert_eq!(
867 ids,
868 vec!["c1", "c2", "c3", "c4", "c5", "c6", "c7", "c8", "c9"]
869 );
870 assert_eq!(cmds[0].causation_id(), Some("root"));
871 assert_eq!(cmds[1].causation_id(), None);
872 }
873
874 #[test]
875 fn canonical_route_spec_serde_roundtrip() {
876 let mut spec = CanonicalRouteSpec::new("test-route", "timer:tick?period=1000");
877 spec.steps.push(CanonicalStepSpec::Log {
878 message: "Hello".into(),
879 });
880 spec.steps.push(CanonicalStepSpec::To {
881 uri: "log:info".into(),
882 });
883 spec.steps.push(CanonicalStepSpec::Stop);
884
885 let json = serde_json::to_string(&spec).unwrap();
886 let deserialized: CanonicalRouteSpec = serde_json::from_str(&json).unwrap();
887 assert_eq!(spec, deserialized);
888 }
889
890 #[test]
891 fn canonical_step_spec_serde_variants() {
892 let steps = vec![
893 CanonicalStepSpec::To {
894 uri: "direct:a".into(),
895 },
896 CanonicalStepSpec::Log {
897 message: "msg".into(),
898 },
899 CanonicalStepSpec::WireTap {
900 uri: "direct:audit".into(),
901 },
902 CanonicalStepSpec::Stop,
903 CanonicalStepSpec::Delay {
904 delay_ms: 100,
905 dynamic_header: None,
906 },
907 ];
908 let json = serde_json::to_string_pretty(&steps).unwrap();
909 let back: Vec<CanonicalStepSpec> = serde_json::from_str(&json).unwrap();
910 assert_eq!(steps, back);
911 }
912
913 #[test]
914 fn canonical_route_spec_json_schema_generates() {
915 let schema = schemars::schema_for!(CanonicalRouteSpec);
916 let json = serde_json::to_string(&schema).unwrap();
917 assert!(json.contains("CanonicalRouteSpec"));
918 assert!(json.contains("route_id"));
919 }
920
921 #[test]
922 fn canonical_json_schema_has_no_function_step() {
923 let schema = schemars::schema_for!(CanonicalRouteSpec);
924 let json = serde_json::to_string(&schema).unwrap();
925 assert!(
926 !json.contains("\"function\""),
927 "canonical JSON schema must not contain 'function' step"
928 );
929 }
930
931 #[test]
932 fn canonical_contract_does_not_support_function() {
933 assert!(
934 !canonical_contract_supports_step("function"),
935 "function must not be in CANONICAL_CONTRACT_SUPPORTED_STEPS"
936 );
937 }
938
939 #[test]
940 fn runtime_command_result_all_variants_are_distinct() {
941 let accepted = RuntimeCommandResult::Accepted;
942 let dup = RuntimeCommandResult::Duplicate {
943 command_id: "c1".into(),
944 };
945 let registered = RuntimeCommandResult::RouteRegistered {
946 route_id: "r1".into(),
947 };
948 let changed = RuntimeCommandResult::RouteStateChanged {
949 route_id: "r1".into(),
950 status: "Started".into(),
951 };
952
953 assert_ne!(accepted, dup);
954 assert_ne!(dup, registered);
955 assert_ne!(registered, changed);
956
957 let dup2 = RuntimeCommandResult::Duplicate {
958 command_id: "c1".into(),
959 };
960 assert_eq!(dup, dup2);
961 }
962
963 #[test]
964 fn runtime_event_serialization_round_trip() {
965 let event = RuntimeEvent::RouteFailed {
966 route_id: "route-a".to_string(),
967 error: "boom".to_string(),
968 };
969 let json = serde_json::to_string(&event).unwrap();
970 let back: RuntimeEvent = serde_json::from_str(&json).unwrap();
971 assert_eq!(event, back);
972 }
973
974 #[test]
975 fn noop_runtime_execute_and_ask_return_expected_shapes() {
976 let rt = NoopRuntime;
977 let cmd = RuntimeCommand::RegisterRoute {
978 spec: CanonicalRouteSpec::new("r2", "timer:tick"),
979 command_id: "c1".into(),
980 causation_id: None,
981 };
982 let cmd_result = block_on(rt.execute(cmd)).unwrap();
983 assert_eq!(
984 cmd_result,
985 RuntimeCommandResult::RouteRegistered {
986 route_id: "r2".into()
987 }
988 );
989
990 let query_result = block_on(rt.ask(RuntimeQuery::GetRouteStatus {
991 route_id: "r2".into(),
992 }))
993 .unwrap();
994 assert_eq!(
995 query_result,
996 RuntimeQueryResult::RouteStatus {
997 route_id: "r2".into(),
998 status: "Started".into()
999 }
1000 );
1001 }
1002
1003 #[test]
1004 fn canonical_contract_name_and_version_constants_match() {
1005 assert_eq!(CANONICAL_CONTRACT_NAME, "canonical-v1");
1006 assert_eq!(CANONICAL_CONTRACT_VERSION, 2);
1007 }
1008
1009 #[test]
1010 fn canonical_concurrency_spec_rejects_zero_max() {
1011 let spec = CanonicalRouteSpec::new("r1", "timer:tick")
1012 .with_concurrency(CanonicalConcurrencySpec::Concurrent { max: 0 });
1013 let err = spec.validate_contract().unwrap_err().to_string();
1014 assert!(err.contains("concurrency max must be > 0"), "{err}");
1015 }
1016
1017 #[test]
1018 fn canonical_v2_round_trip() {
1019 let spec = CanonicalRouteSpec::new("r1", "timer:tick")
1020 .with_auto_startup(false)
1021 .with_startup_order(42)
1022 .with_concurrency(CanonicalConcurrencySpec::Concurrent { max: 8 });
1023 spec.validate_contract().unwrap();
1024 }
1025
1026 #[test]
1027 fn canonical_v2_version_is_2() {
1028 assert_eq!(CANONICAL_CONTRACT_VERSION, 2);
1029 }
1030
1031 #[test]
1032 fn canonical_loss_report_builder() {
1033 let report =
1034 CanonicalLossReport::from_field("error_handler", "not supported by canonical path", 2);
1035 assert_eq!(report.dropped_fields.len(), 1);
1036 assert_eq!(report.dropped_fields[0].field, "error_handler");
1037 }
1038
1039 #[test]
1040 fn canonical_v1_json_deserializes_in_v2() {
1041 let json = r#"{"route_id":"r1","from":"timer:tick","steps":[],"version":1}"#;
1042 let spec: CanonicalRouteSpec = serde_json::from_str(json).unwrap();
1043 assert_eq!(spec.route_id, "r1");
1044 assert!(spec.auto_startup.is_none());
1045 assert!(spec.startup_order.is_none());
1046 assert!(spec.concurrency.is_none());
1047 spec.validate_contract().unwrap();
1049 }
1050}