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