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