1use std::collections::{BTreeMap, BTreeSet};
7use std::time::Duration;
8
9use nasaga_core::{
10 CancelMode, Compensation, DefinitionVersion, ResolutionMode, ResolutionSpec, ServiceIdentity,
11 StepDefinition, StepName, TimeoutPolicy, WorkflowDefinition, WorkflowName,
12 OPAQUE_IDENTIFIER_MAX_BYTES, STRUCTURED_IDENTIFIER_MAX_BYTES,
13};
14use serde::{Deserialize, Serialize};
15use sha2::{Digest as _, Sha256};
16use url::Url;
17
18use crate::DefinitionRegistry;
19
20pub const CATALOG_OBJECT_KEY_MAX_LEN: usize =
22 OPAQUE_IDENTIFIER_MAX_BYTES + STRUCTURED_IDENTIFIER_MAX_BYTES * 4 + 10 + 5;
23
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
26#[serde(rename_all = "snake_case")]
27pub enum DefinitionLifecycle {
28 Candidate,
30 Active,
32 Deprecated,
34 Retired,
36}
37
38impl DefinitionLifecycle {
39 pub fn as_str(self) -> &'static str {
45 match self {
46 Self::Candidate => "candidate",
47 Self::Active => "active",
48 Self::Deprecated => "deprecated",
49 Self::Retired => "retired",
50 }
51 }
52}
53
54#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
56#[serde(rename_all = "snake_case")]
57pub enum DefinitionPublishDisposition {
58 Published,
60 Duplicate,
62}
63
64#[derive(Debug, Clone, Copy, PartialEq, Eq)]
66#[non_exhaustive]
67pub enum DefinitionCatalogError {
68 InvalidArgument,
70 PermissionDenied,
72 NotFound,
74 FailedPrecondition,
76 DigestConflict,
78 PreconditionFailed,
80 OperationConflict,
82}
83
84impl DefinitionCatalogError {
85 pub fn from_error(error: &anyhow::Error) -> Option<Self> {
92 error
93 .chain()
94 .find_map(|cause| cause.downcast_ref::<Self>().copied())
95 }
96}
97
98impl std::fmt::Display for DefinitionCatalogError {
99 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
106 formatter.write_str(match self {
107 Self::InvalidArgument => "Saga Catalog request is invalid",
108 Self::PermissionDenied => "Saga Catalog actor is not authorized for this object",
109 Self::NotFound => "Saga definition was not found",
110 Self::FailedPrecondition => "Saga Catalog business precondition is not satisfied",
111 Self::DigestConflict => "Saga definition key has a different immutable digest",
112 Self::PreconditionFailed => "Saga definition digest no longer matches",
113 Self::OperationConflict => {
114 "Saga definition operation identity has different parameters"
115 }
116 })
117 }
118}
119
120impl std::error::Error for DefinitionCatalogError {}
121
122#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
124#[serde(rename_all = "kebab-case")]
125pub enum DefinitionCompensation {
126 Compensable,
128 NonCompensable,
130}
131
132impl From<DefinitionCompensation> for Compensation {
133 fn from(value: DefinitionCompensation) -> Self {
140 match value {
141 DefinitionCompensation::Compensable => Self::Compensable,
142 DefinitionCompensation::NonCompensable => Self::NonCompensable,
143 }
144 }
145}
146
147impl From<Compensation> for DefinitionCompensation {
148 fn from(value: Compensation) -> Self {
155 match value {
156 Compensation::Compensable => Self::Compensable,
157 Compensation::NonCompensable => Self::NonCompensable,
158 }
159 }
160}
161
162#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
164#[serde(rename_all = "kebab-case")]
165pub enum DefinitionCancelMode {
166 LocalFenceable,
168 ExternallyCancellable,
170 ResolveOnly,
172}
173
174impl From<DefinitionCancelMode> for CancelMode {
175 fn from(value: DefinitionCancelMode) -> Self {
182 match value {
183 DefinitionCancelMode::LocalFenceable => Self::LocalFenceable,
184 DefinitionCancelMode::ExternallyCancellable => Self::ExternallyCancellable,
185 DefinitionCancelMode::ResolveOnly => Self::ResolveOnly,
186 }
187 }
188}
189
190impl From<CancelMode> for DefinitionCancelMode {
191 fn from(value: CancelMode) -> Self {
198 match value {
199 CancelMode::LocalFenceable => Self::LocalFenceable,
200 CancelMode::ExternallyCancellable => Self::ExternallyCancellable,
201 CancelMode::ResolveOnly => Self::ResolveOnly,
202 }
203 }
204}
205
206#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
208#[serde(rename_all = "snake_case")]
209pub enum DefinitionResolutionMode {
210 Callback,
212 Poll,
214 Manual,
216}
217
218impl From<DefinitionResolutionMode> for ResolutionMode {
219 fn from(value: DefinitionResolutionMode) -> Self {
226 match value {
227 DefinitionResolutionMode::Callback => Self::Callback,
228 DefinitionResolutionMode::Poll => Self::Poll,
229 DefinitionResolutionMode::Manual => Self::Manual,
230 }
231 }
232}
233
234impl From<ResolutionMode> for DefinitionResolutionMode {
235 fn from(value: ResolutionMode) -> Self {
242 match value {
243 ResolutionMode::Callback => Self::Callback,
244 ResolutionMode::Poll => Self::Poll,
245 ResolutionMode::Manual => Self::Manual,
246 }
247 }
248}
249
250#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
252#[serde(untagged)]
253pub enum DefinitionResolution {
254 Forbidden(String),
256 Allowed {
258 mode: DefinitionResolutionMode,
260 budget_ms: u64,
262 },
263}
264
265#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
267#[serde(rename_all = "kebab-case")]
268pub enum DefinitionTimeoutPolicy {
269 AcceptLateSuccess,
271 CompensateLateSuccess,
273}
274
275impl From<DefinitionTimeoutPolicy> for TimeoutPolicy {
276 fn from(value: DefinitionTimeoutPolicy) -> Self {
283 match value {
284 DefinitionTimeoutPolicy::AcceptLateSuccess => Self::AcceptLateSuccess,
285 DefinitionTimeoutPolicy::CompensateLateSuccess => Self::CompensateLateSuccess,
286 }
287 }
288}
289
290impl From<TimeoutPolicy> for DefinitionTimeoutPolicy {
291 fn from(value: TimeoutPolicy) -> Self {
298 match value {
299 TimeoutPolicy::AcceptLateSuccess => Self::AcceptLateSuccess,
300 TimeoutPolicy::CompensateLateSuccess => Self::CompensateLateSuccess,
301 }
302 }
303}
304
305#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
307#[serde(deny_unknown_fields)]
308pub struct DefinitionStepArtifact {
309 #[serde(default, skip_serializing_if = "Option::is_none")]
311 pub payload_contract: Option<nasaga_core::SagaPayloadContract>,
312 pub name: String,
314 pub owner: String,
316 pub compensation: DefinitionCompensation,
318 pub cancel_mode: DefinitionCancelMode,
320 pub resolution: DefinitionResolution,
322 pub timeout_ms: u64,
324 pub timeout_policy: DefinitionTimeoutPolicy,
326}
327
328#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
330#[serde(deny_unknown_fields)]
331pub struct DefinitionArtifact {
332 pub tenant: String,
334 pub workflow_owner: String,
336 pub workflow: String,
338 pub definition_version: u32,
340 pub steps: Vec<DefinitionStepArtifact>,
342 pub seal: String,
344}
345
346impl DefinitionArtifact {
347 pub fn to_definition(&self) -> anyhow::Result<WorkflowDefinition> {
353 let workflow =
354 WorkflowName::new(&self.workflow).map_err(|error| anyhow::anyhow!(error.code()))?;
355 let version = DefinitionVersion::new(self.definition_version)
356 .map_err(|error| anyhow::anyhow!(error.code()))?;
357 let mut steps = Vec::with_capacity(self.steps.len());
358 for artifact in &self.steps {
359 let resolution = match &artifact.resolution {
360 DefinitionResolution::Forbidden(value) if value == "forbidden" => {
361 ResolutionSpec::forbidden()
362 }
363 DefinitionResolution::Forbidden(_) => {
364 anyhow::bail!("definition resolution text must be `forbidden`");
365 }
366 DefinitionResolution::Allowed {
367 mode,
368 budget_ms,
369 } => ResolutionSpec::allowed((*mode).into(), Duration::from_millis(*budget_ms)),
370 };
371 steps.push(
372 StepDefinition::new(
373 StepName::new(&artifact.name).map_err(|error| anyhow::anyhow!(error.code()))?,
374 ServiceIdentity::new(&artifact.owner)
375 .map_err(|error| anyhow::anyhow!(error.code()))?,
376 artifact.compensation.into(),
377 artifact.cancel_mode.into(),
378 resolution,
379 Duration::from_millis(artifact.timeout_ms),
380 artifact.timeout_policy.into(),
381 )
382 .with_payload_contract(artifact.payload_contract.clone().unwrap_or_default())?,
383 );
384 }
385 let definition = WorkflowDefinition::new(workflow, version, steps)
386 .map_err(|error| anyhow::anyhow!(error.code()))?;
387 if definition.digest() != self.seal {
388 anyhow::bail!("definition seal does not match canonical workflow content");
389 }
390 Ok(definition)
391 }
392
393 pub fn from_definition(
402 tenant: impl Into<String>,
403 workflow_owner: impl Into<String>,
404 definition: &WorkflowDefinition,
405 ) -> Self {
406 let steps = definition
407 .steps()
408 .iter()
409 .map(|step| DefinitionStepArtifact {
410 payload_contract: (step.payload_contract()
411 != &nasaga_core::SagaPayloadContract::default())
412 .then(|| step.payload_contract().clone()),
413 name: step.name().as_str().to_owned(),
414 owner: step.owner().as_str().to_owned(),
415 compensation: step.compensation().into(),
416 cancel_mode: step.cancel_mode().into(),
417 resolution: match step.resolution().mode() {
418 Some(mode) => DefinitionResolution::Allowed {
419 mode: mode.into(),
420 budget_ms: u64::try_from(step.resolution().budget().as_millis())
421 .unwrap_or(u64::MAX),
422 },
423 None => DefinitionResolution::Forbidden("forbidden".to_owned()),
424 },
425 timeout_ms: u64::try_from(step.timeout().as_millis()).unwrap_or(u64::MAX),
426 timeout_policy: step.timeout_policy().into(),
427 })
428 .collect();
429 Self {
430 tenant: tenant.into(),
431 workflow_owner: workflow_owner.into(),
432 workflow: definition.name().as_str().to_owned(),
433 definition_version: definition.version().get(),
434 steps,
435 seal: definition.digest(),
436 }
437 }
438}
439
440#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
442#[serde(deny_unknown_fields)]
443pub struct DefinitionRecord {
444 pub artifact: DefinitionArtifact,
446 pub lifecycle: DefinitionLifecycle,
448 pub catalog_generation: u64,
450 pub published_at_ms: i64,
452 pub activated_at_ms: Option<i64>,
454}
455
456#[derive(Debug, Clone, PartialEq, Eq)]
458pub struct DefinitionActivationGate {
459 orchestrator_service_identity: ServiceIdentity,
460 catalog_generation: u64,
461 snapshot_digest: String,
462 transport: String,
463 redis_key_tag: Option<String>,
464 address_policy_digest: Option<String>,
465 publisher_contract_digest: Option<String>,
466 result_backend_digest: Option<String>,
467 result_contract_digests: BTreeMap<String, BTreeSet<String>>,
468 activation_contract_digest: String,
469}
470
471#[derive(Debug, Clone, PartialEq, Eq)]
473pub struct DefinitionLifecycleOperation {
474 expected_sha256: String,
475 operation_id: String,
476 reason: String,
477}
478
479impl DefinitionLifecycleOperation {
480 pub fn new(
489 expected_sha256: impl Into<String>,
490 operation_id: impl Into<String>,
491 reason: impl Into<String>,
492 ) -> anyhow::Result<Self> {
493 let expected_sha256 = expected_sha256.into();
494 let operation_id = operation_id.into();
495 let reason = reason.into();
496 anyhow::ensure!(
497 expected_sha256.len() == 64
498 && expected_sha256.bytes().all(|byte| byte.is_ascii_hexdigit()),
499 "definition expected digest is invalid"
500 );
501 anyhow::ensure!(
502 !operation_id.is_empty()
503 && operation_id == operation_id.trim()
504 && operation_id.len() <= 190
505 && !operation_id.chars().any(char::is_control),
506 "definition operation identity is invalid"
507 );
508 anyhow::ensure!(
509 !reason.is_empty()
510 && reason == reason.trim()
511 && reason.len() <= 512
512 && !reason.chars().any(char::is_control),
513 "definition operation reason is invalid"
514 );
515 Ok(Self {
516 expected_sha256: expected_sha256.to_ascii_lowercase(),
517 operation_id,
518 reason,
519 })
520 }
521
522 pub fn for_validated_activation(
531 artifact: &DefinitionArtifact,
532 catalog_generation: u64,
533 reason: impl Into<String>,
534 ) -> anyhow::Result<Self> {
535 let mut canonical = Vec::new();
536 for field in [
537 b"nasaga-validated-activation".as_slice(),
538 artifact.seal.as_bytes(),
539 artifact.tenant.as_bytes(),
540 artifact.workflow.as_bytes(),
541 ] {
542 canonical.extend_from_slice(&(field.len() as u64).to_be_bytes());
543 canonical.extend_from_slice(field);
544 }
545 canonical.extend_from_slice(&artifact.definition_version.to_be_bytes());
546 canonical.extend_from_slice(&catalog_generation.to_be_bytes());
547 let mut hasher = Sha256::new();
548 hasher.update(canonical);
549 Self::new(
550 artifact.seal.clone(),
551 format!("validated-{}", hex::encode(hasher.finalize())),
552 reason,
553 )
554 }
555
556 pub fn expected_sha256(&self) -> &str {
562 &self.expected_sha256
563 }
564
565 pub fn operation_id(&self) -> &str {
571 &self.operation_id
572 }
573
574 pub fn reason(&self) -> &str {
580 &self.reason
581 }
582}
583
584impl DefinitionActivationGate {
585 pub fn new(
595 orchestrator_service_identity: ServiceIdentity,
596 catalog_generation: u64,
597 snapshot_digest: impl Into<String>,
598 transport: impl Into<String>,
599 ) -> anyhow::Result<Self> {
600 let snapshot_digest = snapshot_digest.into();
601 anyhow::ensure!(
602 snapshot_digest.len() == 64
603 && snapshot_digest.bytes().all(|byte| byte.is_ascii_hexdigit()),
604 "catalog snapshot digest is invalid"
605 );
606 let transport = transport.into();
607 anyhow::ensure!(
608 matches!(
609 transport.as_str(),
610 "http" | "grpc" | "kafka" | "redis-stream"
611 ),
612 "definition activation transport is unsupported"
613 );
614 let mut gate = Self {
615 orchestrator_service_identity,
616 catalog_generation,
617 snapshot_digest,
618 transport,
619 redis_key_tag: None,
620 address_policy_digest: None,
621 publisher_contract_digest: None,
622 result_backend_digest: None,
623 result_contract_digests: BTreeMap::new(),
624 activation_contract_digest: String::new(),
625 };
626 gate.refresh_activation_contract_digest();
627 Ok(gate)
628 }
629
630 pub fn with_redis_key_tag(mut self, key_tag: impl Into<String>) -> anyhow::Result<Self> {
638 let key_tag = key_tag.into();
639 anyhow::ensure!(
640 self.transport == "redis-stream" && valid_redis_key_tag(&key_tag),
641 "definition activation Redis key tag is invalid"
642 );
643 self.redis_key_tag = Some(key_tag);
644 self.refresh_activation_contract_digest();
645 Ok(self)
646 }
647
648 pub fn with_address_policy_digest(mut self, digest: impl Into<String>) -> anyhow::Result<Self> {
655 let digest = digest.into();
656 anyhow::ensure!(
657 digest.len() == 64 && digest.bytes().all(|byte| byte.is_ascii_hexdigit()),
658 "definition activation address policy digest is invalid"
659 );
660 self.address_policy_digest = Some(digest.to_ascii_lowercase());
661 self.refresh_activation_contract_digest();
662 Ok(self)
663 }
664
665 pub fn with_publisher_contract_digest(
673 mut self,
674 digest: impl Into<String>,
675 ) -> anyhow::Result<Self> {
676 let digest = digest.into();
677 anyhow::ensure!(
678 digest.len() == 64 && digest.bytes().all(|byte| byte.is_ascii_hexdigit()),
679 "definition activation publisher contract digest is invalid"
680 );
681 self.publisher_contract_digest = Some(digest.to_ascii_lowercase());
682 self.refresh_activation_contract_digest();
683 Ok(self)
684 }
685
686 pub fn with_result_backend_digest(mut self, digest: impl Into<String>) -> anyhow::Result<Self> {
694 let digest = digest.into();
695 anyhow::ensure!(
696 self.transport == "kafka"
697 && digest.len() == 64
698 && digest.bytes().all(|byte| byte.is_ascii_hexdigit()),
699 "definition activation result backend digest is invalid"
700 );
701 self.result_backend_digest = Some(digest.to_ascii_lowercase());
702 self.refresh_activation_contract_digest();
703 Ok(self)
704 }
705
706 pub fn with_result_contract_digest(
714 mut self,
715 owner: impl Into<String>,
716 digest: impl Into<String>,
717 ) -> anyhow::Result<Self> {
718 let owner = owner.into();
719 ServiceIdentity::new(&owner)?;
720 let digest = digest.into();
721 anyhow::ensure!(
722 digest.len() == 64 && digest.bytes().all(|byte| byte.is_ascii_hexdigit()),
723 "definition activation result contract digest is invalid"
724 );
725 self.result_contract_digests
726 .entry(owner)
727 .or_default()
728 .insert(digest.to_ascii_lowercase());
729 self.refresh_activation_contract_digest();
730 Ok(self)
731 }
732
733 fn refresh_activation_contract_digest(&mut self) {
739 let mut digest = Sha256::new();
740 update_snapshot_digest(&mut digest, b"nasaga-activation-contract");
741 update_snapshot_digest(&mut digest, self.transport.as_bytes());
742 match self.redis_key_tag.as_deref() {
743 Some(key_tag) => {
744 digest.update([1]);
745 update_snapshot_digest(&mut digest, key_tag.as_bytes());
746 }
747 None => digest.update([0]),
748 }
749 match self.address_policy_digest.as_deref() {
750 Some(policy_digest) => {
751 digest.update([1]);
752 update_snapshot_digest(&mut digest, policy_digest.as_bytes());
753 }
754 None => digest.update([0]),
755 }
756 match self.publisher_contract_digest.as_deref() {
757 Some(publisher_digest) => {
758 digest.update([1]);
759 update_snapshot_digest(&mut digest, publisher_digest.as_bytes());
760 }
761 None => digest.update([0]),
762 }
763 match self.result_backend_digest.as_deref() {
764 Some(result_backend_digest) => {
765 digest.update([1]);
766 update_snapshot_digest(&mut digest, result_backend_digest.as_bytes());
767 }
768 None => digest.update([0]),
769 }
770 digest.update((self.result_contract_digests.len() as u64).to_be_bytes());
771 for (owner, contracts) in &self.result_contract_digests {
772 update_snapshot_digest(&mut digest, owner.as_bytes());
773 digest.update((contracts.len() as u64).to_be_bytes());
774 for contract in contracts {
775 update_snapshot_digest(&mut digest, contract.as_bytes());
776 }
777 }
778 self.activation_contract_digest = hex::encode(digest.finalize());
779 }
780
781 pub fn orchestrator_service_identity(&self) -> &ServiceIdentity {
787 &self.orchestrator_service_identity
788 }
789
790 pub fn catalog_generation(&self) -> u64 {
796 self.catalog_generation
797 }
798
799 pub fn snapshot_digest(&self) -> &str {
805 &self.snapshot_digest
806 }
807
808 pub fn transport(&self) -> &str {
814 &self.transport
815 }
816
817 pub fn redis_key_tag(&self) -> Option<&str> {
823 self.redis_key_tag.as_deref()
824 }
825
826 pub fn result_backend_digest(&self) -> Option<&str> {
832 self.result_backend_digest.as_deref()
833 }
834
835 pub fn accepts_result_contract(&self, owner: &str, digest: &str) -> bool {
843 self.result_contract_digests
844 .get(owner)
845 .is_some_and(|contracts| contracts.contains(digest))
846 }
847
848 pub fn activation_contract_digest(&self) -> &str {
854 &self.activation_contract_digest
855 }
856}
857
858#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
860#[serde(deny_unknown_fields)]
861pub struct CapabilityDescriptor {
862 #[serde(default, skip_serializing_if = "Option::is_none")]
864 pub payload_contract: Option<nasaga_core::SagaPayloadContract>,
865 pub tenant: String,
867 pub owner: String,
869 pub replica_identity: String,
871 pub workflow: String,
873 pub definition_version: u32,
875 pub step: String,
877 pub compensation: DefinitionCompensation,
879 pub cancel_mode: DefinitionCancelMode,
881 pub allow_unknown: bool,
883 pub resolution_mode: Option<DefinitionResolutionMode>,
885 pub transport: String,
887 pub endpoint: String,
889 pub effective_saga_base_path: Option<String>,
891 #[serde(default, skip_serializing_if = "Option::is_none")]
893 pub result_contract_digest: Option<String>,
894 pub route_generation: u64,
896 pub requested_lease_ms: u64,
898}
899
900impl CapabilityDescriptor {
901 pub fn digest(&self) -> String {
907 let mut canonical = self.clone();
908 canonical.requested_lease_ms = 0;
909 let bytes = serde_json::to_vec(&canonical)
910 .expect("serializing a capability descriptor cannot fail");
911 let mut hasher = Sha256::new();
912 hasher.update(bytes);
913 hex::encode(hasher.finalize())
914 }
915
916 pub fn route_contract_digest(&self) -> String {
922 let mut canonical = self.clone();
923 canonical.route_generation = 0;
924 canonical.requested_lease_ms = 0;
925 let bytes = serde_json::to_vec(&canonical)
926 .expect("serializing a capability descriptor cannot fail");
927 let mut hasher = Sha256::new();
928 hasher.update(bytes);
929 hex::encode(hasher.finalize())
930 }
931
932 pub fn matches_step(
941 &self,
942 tenant: &str,
943 definition: &WorkflowDefinition,
944 step: &StepDefinition,
945 ) -> bool {
946 self.payload_contract.clone().unwrap_or_default() == *step.payload_contract()
947 && self.tenant == tenant
948 && self.workflow == definition.name().as_str()
949 && self.definition_version == definition.version().get()
950 && self.step == step.name().as_str()
951 && self.owner == step.owner().as_str()
952 && self.compensation == step.compensation().into()
953 && self.cancel_mode == step.cancel_mode().into()
954 && self.allow_unknown == step.resolution().allow_unknown()
955 && self.resolution_mode == step.resolution().mode().map(Into::into)
956 }
957}
958
959pub fn validate_capability_route_contract(
968 descriptor: &CapabilityDescriptor,
969) -> Result<(), DefinitionCatalogError> {
970 descriptor
971 .payload_contract
972 .clone()
973 .unwrap_or_default()
974 .validate()
975 .map_err(|_| DefinitionCatalogError::InvalidArgument)?;
976 let path_contract_is_valid = match descriptor.transport.as_str() {
977 "http" => descriptor
978 .effective_saga_base_path
979 .as_deref()
980 .is_some_and(valid_capability_http_base_path),
981 "grpc" | "kafka" | "redis-stream" => descriptor.effective_saga_base_path.is_none(),
982 _ => false,
983 };
984 if !path_contract_is_valid {
985 return Err(DefinitionCatalogError::InvalidArgument);
986 }
987 let result_contract_is_valid = match descriptor.transport.as_str() {
988 "http" | "grpc" | "kafka" | "redis-stream" => descriptor
989 .result_contract_digest
990 .as_deref()
991 .is_some_and(|digest| {
992 digest.len() == 64
993 && digest == digest.to_ascii_lowercase()
994 && digest.bytes().all(|byte| byte.is_ascii_hexdigit())
995 }),
996 _ => false,
997 };
998 if !result_contract_is_valid {
999 return Err(DefinitionCatalogError::InvalidArgument);
1000 }
1001
1002 let endpoint_is_valid = match descriptor.transport.as_str() {
1003 "http" => valid_capability_origin(&descriptor.endpoint, &["http", "https"]),
1004 "grpc" => valid_capability_origin(&descriptor.endpoint, &["https"]),
1005 "kafka" => {
1006 !descriptor.endpoint.is_empty()
1007 && descriptor.endpoint.len() <= 249
1008 && descriptor.endpoint.trim() == descriptor.endpoint
1009 && !matches!(descriptor.endpoint.as_str(), "." | "..")
1010 && descriptor
1011 .endpoint
1012 .bytes()
1013 .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
1014 }
1015 "redis-stream" => validate_redis_stream_route(&descriptor.endpoint, None).is_ok(),
1016 _ => false,
1017 };
1018 if endpoint_is_valid {
1019 Ok(())
1020 } else {
1021 Err(DefinitionCatalogError::InvalidArgument)
1022 }
1023}
1024
1025fn valid_capability_origin(endpoint: &str, schemes: &[&str]) -> bool {
1033 let raw_origin_is_canonical = endpoint
1034 .split_once("://")
1035 .is_some_and(|(scheme, authority)| {
1036 schemes.contains(&scheme)
1037 && !authority.is_empty()
1038 && (authority.bytes().all(|byte| byte != b'/')
1039 || (authority.ends_with('/')
1040 && authority[..authority.len() - 1]
1041 .bytes()
1042 .all(|byte| byte != b'/')))
1043 });
1044 Url::parse(endpoint).is_ok_and(|url| {
1045 endpoint.trim() == endpoint
1046 && raw_origin_is_canonical
1047 && schemes.contains(&url.scheme())
1048 && url.host_str().is_some_and(|host| !host.is_empty())
1049 && url.username().is_empty()
1050 && url.password().is_none()
1051 && url.query().is_none()
1052 && url.fragment().is_none()
1053 && matches!(url.path(), "" | "/")
1054 })
1055}
1056
1057pub fn validate_redis_stream_route(
1065 stream: &str,
1066 key_tag: Option<&str>,
1067) -> Result<(), DefinitionCatalogError> {
1068 let valid = !stream.is_empty()
1069 && stream.len() <= 190
1070 && !stream.chars().any(char::is_control)
1071 && key_tag.is_none_or(|tag| {
1072 valid_redis_key_tag(tag) && first_redis_hash_tag(stream) == Some(tag)
1073 });
1074 if valid {
1075 Ok(())
1076 } else {
1077 Err(DefinitionCatalogError::InvalidArgument)
1078 }
1079}
1080
1081fn valid_redis_key_tag(tag: &str) -> bool {
1088 !tag.is_empty()
1089 && tag.len() <= 64
1090 && !tag.contains(['{', '}'])
1091 && !tag.chars().any(char::is_control)
1092}
1093
1094fn first_redis_hash_tag(name: &str) -> Option<&str> {
1102 let open = name.find('{')?;
1103 let close = name[open + 1..].find('}')?;
1104 if close == 0 {
1105 return None;
1106 }
1107 Some(&name[open + 1..open + 1 + close])
1108}
1109
1110fn valid_capability_http_base_path(path: &str) -> bool {
1117 path != "/"
1118 && path.starts_with('/')
1119 && !path.ends_with('/')
1120 && path.bytes().all(|byte| {
1121 byte.is_ascii_alphanumeric() || matches!(byte, b'/' | b'-' | b'_' | b'.' | b'~')
1122 })
1123 && !path.contains("//")
1124 && !path.contains('%')
1125 && !path.contains('?')
1126 && !path.contains('#')
1127 && !path.contains('*')
1128 && !path.contains('{')
1129 && !path.contains('}')
1130 && !path.split('/').any(|segment| matches!(segment, "." | ".."))
1131}
1132
1133#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1135#[serde(deny_unknown_fields)]
1136pub struct CapabilityReceipt {
1137 pub accepted_until_ms: i64,
1139 pub capability_digest: String,
1141 pub route_generation: u64,
1143 pub catalog_generation: u64,
1145}
1146
1147#[derive(Debug, Clone, PartialEq, Eq)]
1149pub struct RegisteredCapability {
1150 pub descriptor: CapabilityDescriptor,
1152 pub accepted_until_ms: i64,
1154 pub capability_digest: String,
1156}
1157
1158pub fn select_capability_route<'a>(
1171 capabilities: &'a [RegisteredCapability],
1172 transport: &str,
1173 tenant: &str,
1174 definition: &WorkflowDefinition,
1175 step: &StepDefinition,
1176) -> Result<Option<&'a CapabilityDescriptor>, DefinitionCatalogError> {
1177 let mut selected: Option<&CapabilityDescriptor> = None;
1178 for descriptor in capabilities
1179 .iter()
1180 .map(|capability| &capability.descriptor)
1181 .filter(|descriptor| {
1182 descriptor.transport == transport && descriptor.matches_step(tenant, definition, step)
1183 })
1184 {
1185 validate_capability_route_contract(descriptor)?;
1186 if selected.is_some_and(|current| {
1187 current.endpoint != descriptor.endpoint
1188 || current.effective_saga_base_path != descriptor.effective_saga_base_path
1189 || (transport == "kafka"
1190 && current.result_contract_digest != descriptor.result_contract_digest)
1191 }) {
1192 return Err(DefinitionCatalogError::FailedPrecondition);
1193 }
1194 selected.get_or_insert(descriptor);
1195 }
1196 Ok(selected)
1197}
1198
1199pub fn select_capability_routes<'a>(
1210 capabilities: &'a [RegisteredCapability],
1211 transport: &str,
1212 tenant: &str,
1213 definition: &WorkflowDefinition,
1214 step: &StepDefinition,
1215) -> Result<Vec<&'a CapabilityDescriptor>, DefinitionCatalogError> {
1216 if !matches!(transport, "http" | "grpc") {
1217 return select_capability_route(capabilities, transport, tenant, definition, step)
1218 .map(|route| route.into_iter().collect());
1219 }
1220 let mut routes: Vec<&CapabilityDescriptor> = Vec::new();
1221 for descriptor in capabilities
1222 .iter()
1223 .map(|capability| &capability.descriptor)
1224 .filter(|descriptor| {
1225 descriptor.transport == transport && descriptor.matches_step(tenant, definition, step)
1226 })
1227 {
1228 validate_capability_route_contract(descriptor)?;
1229 if routes.iter().any(|current| {
1231 current.replica_identity == descriptor.replica_identity && *current != descriptor
1232 }) {
1233 return Err(DefinitionCatalogError::FailedPrecondition);
1234 }
1235 routes.push(descriptor);
1236 }
1237 routes.sort_by(|left, right| {
1238 (&left.endpoint, &left.replica_identity).cmp(&(&right.endpoint, &right.replica_identity))
1239 });
1240 routes.dedup();
1241 Ok(routes)
1242}
1243
1244#[derive(Debug, Clone)]
1246pub struct DynamicCatalogSnapshot {
1247 pub generation: u64,
1249 pub snapshot_digest: String,
1251 pub registry: DefinitionRegistry,
1253 pub definition_records: Vec<DefinitionRecord>,
1255 pub capabilities: Vec<RegisteredCapability>,
1257}
1258
1259impl DynamicCatalogSnapshot {
1260 pub fn new(
1269 generation: u64,
1270 registry: DefinitionRegistry,
1271 capabilities: Vec<RegisteredCapability>,
1272 ) -> Self {
1273 Self::with_definition_records(generation, registry, Vec::new(), capabilities)
1274 }
1275
1276 pub fn with_definition_records(
1286 generation: u64,
1287 registry: DefinitionRegistry,
1288 mut definition_records: Vec<DefinitionRecord>,
1289 mut capabilities: Vec<RegisteredCapability>,
1290 ) -> Self {
1291 definition_records.sort_by(|left, right| {
1292 (
1293 &left.artifact.tenant,
1294 &left.artifact.workflow,
1295 left.artifact.definition_version,
1296 )
1297 .cmp(&(
1298 &right.artifact.tenant,
1299 &right.artifact.workflow,
1300 right.artifact.definition_version,
1301 ))
1302 });
1303 capabilities.sort_by(|left, right| {
1304 (
1305 &left.descriptor.tenant,
1306 &left.descriptor.owner,
1307 &left.descriptor.replica_identity,
1308 &left.descriptor.workflow,
1309 left.descriptor.definition_version,
1310 &left.descriptor.step,
1311 )
1312 .cmp(&(
1313 &right.descriptor.tenant,
1314 &right.descriptor.owner,
1315 &right.descriptor.replica_identity,
1316 &right.descriptor.workflow,
1317 right.descriptor.definition_version,
1318 &right.descriptor.step,
1319 ))
1320 });
1321 let mut digest = Sha256::new();
1322 digest.update(generation.to_be_bytes());
1323 let mut runtime_definitions = registry
1325 .definitions()
1326 .map(|definition| {
1327 (
1328 definition.name().as_str(),
1329 definition.version().get(),
1330 definition.digest(),
1331 )
1332 })
1333 .collect::<Vec<_>>();
1334 runtime_definitions.sort();
1335 for (workflow, version, definition_digest) in runtime_definitions {
1336 update_snapshot_digest(&mut digest, workflow.as_bytes());
1337 digest.update(version.to_be_bytes());
1338 update_snapshot_digest(&mut digest, definition_digest.as_bytes());
1339 }
1340 for record in &definition_records {
1341 update_snapshot_digest(&mut digest, record.artifact.tenant.as_bytes());
1342 update_snapshot_digest(&mut digest, record.artifact.workflow_owner.as_bytes());
1343 update_snapshot_digest(&mut digest, record.artifact.workflow.as_bytes());
1344 digest.update(record.artifact.definition_version.to_be_bytes());
1345 update_snapshot_digest(&mut digest, record.lifecycle.as_str().as_bytes());
1346 update_snapshot_digest(&mut digest, record.artifact.seal.as_bytes());
1347 digest.update(record.catalog_generation.to_be_bytes());
1348 }
1349 for capability in &capabilities {
1350 update_snapshot_digest(&mut digest, capability.descriptor.tenant.as_bytes());
1351 update_snapshot_digest(&mut digest, capability.descriptor.owner.as_bytes());
1352 update_snapshot_digest(
1353 &mut digest,
1354 capability.descriptor.replica_identity.as_bytes(),
1355 );
1356 update_snapshot_digest(&mut digest, capability.descriptor.workflow.as_bytes());
1357 digest.update(capability.descriptor.definition_version.to_be_bytes());
1358 update_snapshot_digest(&mut digest, capability.descriptor.step.as_bytes());
1359 update_snapshot_digest(&mut digest, capability.capability_digest.as_bytes());
1360 }
1361 Self {
1362 generation,
1363 snapshot_digest: hex::encode(digest.finalize()),
1364 registry,
1365 definition_records,
1366 capabilities,
1367 }
1368 }
1369}
1370
1371fn update_snapshot_digest(digest: &mut Sha256, value: &[u8]) {
1379 digest.update((value.len() as u64).to_be_bytes());
1380 digest.update(value);
1381}