use std::collections::{BTreeMap, BTreeSet};
use std::time::Duration;
use nasaga_core::{
CancelMode, Compensation, DefinitionVersion, ResolutionMode, ResolutionSpec, ServiceIdentity,
StepDefinition, StepName, TimeoutPolicy, WorkflowDefinition, WorkflowName,
OPAQUE_IDENTIFIER_MAX_BYTES, STRUCTURED_IDENTIFIER_MAX_BYTES,
};
use serde::{Deserialize, Serialize};
use sha2::{Digest as _, Sha256};
use url::Url;
use crate::DefinitionRegistry;
pub const CATALOG_OBJECT_KEY_MAX_LEN: usize =
OPAQUE_IDENTIFIER_MAX_BYTES + STRUCTURED_IDENTIFIER_MAX_BYTES * 4 + 10 + 5;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DefinitionLifecycle {
Candidate,
Active,
Deprecated,
Retired,
}
impl DefinitionLifecycle {
pub fn as_str(self) -> &'static str {
match self {
Self::Candidate => "candidate",
Self::Active => "active",
Self::Deprecated => "deprecated",
Self::Retired => "retired",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DefinitionPublishDisposition {
Published,
Duplicate,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum DefinitionCatalogError {
InvalidArgument,
PermissionDenied,
NotFound,
FailedPrecondition,
DigestConflict,
PreconditionFailed,
OperationConflict,
}
impl DefinitionCatalogError {
pub fn from_error(error: &anyhow::Error) -> Option<Self> {
error
.chain()
.find_map(|cause| cause.downcast_ref::<Self>().copied())
}
}
impl std::fmt::Display for DefinitionCatalogError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str(match self {
Self::InvalidArgument => "Saga Catalog request is invalid",
Self::PermissionDenied => "Saga Catalog actor is not authorized for this object",
Self::NotFound => "Saga definition was not found",
Self::FailedPrecondition => "Saga Catalog business precondition is not satisfied",
Self::DigestConflict => "Saga definition key has a different immutable digest",
Self::PreconditionFailed => "Saga definition digest no longer matches",
Self::OperationConflict => {
"Saga definition operation identity has different parameters"
}
})
}
}
impl std::error::Error for DefinitionCatalogError {}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum DefinitionCompensation {
Compensable,
NonCompensable,
}
impl From<DefinitionCompensation> for Compensation {
fn from(value: DefinitionCompensation) -> Self {
match value {
DefinitionCompensation::Compensable => Self::Compensable,
DefinitionCompensation::NonCompensable => Self::NonCompensable,
}
}
}
impl From<Compensation> for DefinitionCompensation {
fn from(value: Compensation) -> Self {
match value {
Compensation::Compensable => Self::Compensable,
Compensation::NonCompensable => Self::NonCompensable,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum DefinitionCancelMode {
LocalFenceable,
ExternallyCancellable,
ResolveOnly,
}
impl From<DefinitionCancelMode> for CancelMode {
fn from(value: DefinitionCancelMode) -> Self {
match value {
DefinitionCancelMode::LocalFenceable => Self::LocalFenceable,
DefinitionCancelMode::ExternallyCancellable => Self::ExternallyCancellable,
DefinitionCancelMode::ResolveOnly => Self::ResolveOnly,
}
}
}
impl From<CancelMode> for DefinitionCancelMode {
fn from(value: CancelMode) -> Self {
match value {
CancelMode::LocalFenceable => Self::LocalFenceable,
CancelMode::ExternallyCancellable => Self::ExternallyCancellable,
CancelMode::ResolveOnly => Self::ResolveOnly,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DefinitionResolutionMode {
Callback,
Poll,
Manual,
}
impl From<DefinitionResolutionMode> for ResolutionMode {
fn from(value: DefinitionResolutionMode) -> Self {
match value {
DefinitionResolutionMode::Callback => Self::Callback,
DefinitionResolutionMode::Poll => Self::Poll,
DefinitionResolutionMode::Manual => Self::Manual,
}
}
}
impl From<ResolutionMode> for DefinitionResolutionMode {
fn from(value: ResolutionMode) -> Self {
match value {
ResolutionMode::Callback => Self::Callback,
ResolutionMode::Poll => Self::Poll,
ResolutionMode::Manual => Self::Manual,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(untagged)]
pub enum DefinitionResolution {
Forbidden(String),
Allowed {
mode: DefinitionResolutionMode,
budget_ms: u64,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum DefinitionTimeoutPolicy {
AcceptLateSuccess,
CompensateLateSuccess,
}
impl From<DefinitionTimeoutPolicy> for TimeoutPolicy {
fn from(value: DefinitionTimeoutPolicy) -> Self {
match value {
DefinitionTimeoutPolicy::AcceptLateSuccess => Self::AcceptLateSuccess,
DefinitionTimeoutPolicy::CompensateLateSuccess => Self::CompensateLateSuccess,
}
}
}
impl From<TimeoutPolicy> for DefinitionTimeoutPolicy {
fn from(value: TimeoutPolicy) -> Self {
match value {
TimeoutPolicy::AcceptLateSuccess => Self::AcceptLateSuccess,
TimeoutPolicy::CompensateLateSuccess => Self::CompensateLateSuccess,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DefinitionStepArtifact {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub payload_contract: Option<nasaga_core::SagaPayloadContract>,
pub name: String,
pub owner: String,
pub compensation: DefinitionCompensation,
pub cancel_mode: DefinitionCancelMode,
pub resolution: DefinitionResolution,
pub timeout_ms: u64,
pub timeout_policy: DefinitionTimeoutPolicy,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DefinitionArtifact {
pub tenant: String,
pub workflow_owner: String,
pub workflow: String,
pub definition_version: u32,
pub steps: Vec<DefinitionStepArtifact>,
pub seal: String,
}
impl DefinitionArtifact {
pub fn to_definition(&self) -> anyhow::Result<WorkflowDefinition> {
let workflow =
WorkflowName::new(&self.workflow).map_err(|error| anyhow::anyhow!(error.code()))?;
let version = DefinitionVersion::new(self.definition_version)
.map_err(|error| anyhow::anyhow!(error.code()))?;
let mut steps = Vec::with_capacity(self.steps.len());
for artifact in &self.steps {
let resolution = match &artifact.resolution {
DefinitionResolution::Forbidden(value) if value == "forbidden" => {
ResolutionSpec::forbidden()
}
DefinitionResolution::Forbidden(_) => {
anyhow::bail!("definition resolution text must be `forbidden`");
}
DefinitionResolution::Allowed {
mode,
budget_ms,
} => ResolutionSpec::allowed((*mode).into(), Duration::from_millis(*budget_ms)),
};
steps.push(
StepDefinition::new(
StepName::new(&artifact.name).map_err(|error| anyhow::anyhow!(error.code()))?,
ServiceIdentity::new(&artifact.owner)
.map_err(|error| anyhow::anyhow!(error.code()))?,
artifact.compensation.into(),
artifact.cancel_mode.into(),
resolution,
Duration::from_millis(artifact.timeout_ms),
artifact.timeout_policy.into(),
)
.with_payload_contract(artifact.payload_contract.clone().unwrap_or_default())?,
);
}
let definition = WorkflowDefinition::new(workflow, version, steps)
.map_err(|error| anyhow::anyhow!(error.code()))?;
if definition.digest() != self.seal {
anyhow::bail!("definition seal does not match canonical workflow content");
}
Ok(definition)
}
pub fn from_definition(
tenant: impl Into<String>,
workflow_owner: impl Into<String>,
definition: &WorkflowDefinition,
) -> Self {
let steps = definition
.steps()
.iter()
.map(|step| DefinitionStepArtifact {
payload_contract: (step.payload_contract()
!= &nasaga_core::SagaPayloadContract::default())
.then(|| step.payload_contract().clone()),
name: step.name().as_str().to_owned(),
owner: step.owner().as_str().to_owned(),
compensation: step.compensation().into(),
cancel_mode: step.cancel_mode().into(),
resolution: match step.resolution().mode() {
Some(mode) => DefinitionResolution::Allowed {
mode: mode.into(),
budget_ms: u64::try_from(step.resolution().budget().as_millis())
.unwrap_or(u64::MAX),
},
None => DefinitionResolution::Forbidden("forbidden".to_owned()),
},
timeout_ms: u64::try_from(step.timeout().as_millis()).unwrap_or(u64::MAX),
timeout_policy: step.timeout_policy().into(),
})
.collect();
Self {
tenant: tenant.into(),
workflow_owner: workflow_owner.into(),
workflow: definition.name().as_str().to_owned(),
definition_version: definition.version().get(),
steps,
seal: definition.digest(),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DefinitionRecord {
pub artifact: DefinitionArtifact,
pub lifecycle: DefinitionLifecycle,
pub catalog_generation: u64,
pub published_at_ms: i64,
pub activated_at_ms: Option<i64>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DefinitionActivationGate {
orchestrator_service_identity: ServiceIdentity,
catalog_generation: u64,
snapshot_digest: String,
transport: String,
redis_key_tag: Option<String>,
address_policy_digest: Option<String>,
publisher_contract_digest: Option<String>,
result_backend_digest: Option<String>,
result_contract_digests: BTreeMap<String, BTreeSet<String>>,
activation_contract_digest: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DefinitionLifecycleOperation {
expected_sha256: String,
operation_id: String,
reason: String,
}
impl DefinitionLifecycleOperation {
pub fn new(
expected_sha256: impl Into<String>,
operation_id: impl Into<String>,
reason: impl Into<String>,
) -> anyhow::Result<Self> {
let expected_sha256 = expected_sha256.into();
let operation_id = operation_id.into();
let reason = reason.into();
anyhow::ensure!(
expected_sha256.len() == 64
&& expected_sha256.bytes().all(|byte| byte.is_ascii_hexdigit()),
"definition expected digest is invalid"
);
anyhow::ensure!(
!operation_id.is_empty()
&& operation_id == operation_id.trim()
&& operation_id.len() <= 190
&& !operation_id.chars().any(char::is_control),
"definition operation identity is invalid"
);
anyhow::ensure!(
!reason.is_empty()
&& reason == reason.trim()
&& reason.len() <= 512
&& !reason.chars().any(char::is_control),
"definition operation reason is invalid"
);
Ok(Self {
expected_sha256: expected_sha256.to_ascii_lowercase(),
operation_id,
reason,
})
}
pub fn for_validated_activation(
artifact: &DefinitionArtifact,
catalog_generation: u64,
reason: impl Into<String>,
) -> anyhow::Result<Self> {
let mut canonical = Vec::new();
for field in [
b"nasaga-validated-activation".as_slice(),
artifact.seal.as_bytes(),
artifact.tenant.as_bytes(),
artifact.workflow.as_bytes(),
] {
canonical.extend_from_slice(&(field.len() as u64).to_be_bytes());
canonical.extend_from_slice(field);
}
canonical.extend_from_slice(&artifact.definition_version.to_be_bytes());
canonical.extend_from_slice(&catalog_generation.to_be_bytes());
let mut hasher = Sha256::new();
hasher.update(canonical);
Self::new(
artifact.seal.clone(),
format!("validated-{}", hex::encode(hasher.finalize())),
reason,
)
}
pub fn expected_sha256(&self) -> &str {
&self.expected_sha256
}
pub fn operation_id(&self) -> &str {
&self.operation_id
}
pub fn reason(&self) -> &str {
&self.reason
}
}
impl DefinitionActivationGate {
pub fn new(
orchestrator_service_identity: ServiceIdentity,
catalog_generation: u64,
snapshot_digest: impl Into<String>,
transport: impl Into<String>,
) -> anyhow::Result<Self> {
let snapshot_digest = snapshot_digest.into();
anyhow::ensure!(
snapshot_digest.len() == 64
&& snapshot_digest.bytes().all(|byte| byte.is_ascii_hexdigit()),
"catalog snapshot digest is invalid"
);
let transport = transport.into();
anyhow::ensure!(
matches!(
transport.as_str(),
"http" | "grpc" | "kafka" | "redis-stream"
),
"definition activation transport is unsupported"
);
let mut gate = Self {
orchestrator_service_identity,
catalog_generation,
snapshot_digest,
transport,
redis_key_tag: None,
address_policy_digest: None,
publisher_contract_digest: None,
result_backend_digest: None,
result_contract_digests: BTreeMap::new(),
activation_contract_digest: String::new(),
};
gate.refresh_activation_contract_digest();
Ok(gate)
}
pub fn with_redis_key_tag(mut self, key_tag: impl Into<String>) -> anyhow::Result<Self> {
let key_tag = key_tag.into();
anyhow::ensure!(
self.transport == "redis-stream" && valid_redis_key_tag(&key_tag),
"definition activation Redis key tag is invalid"
);
self.redis_key_tag = Some(key_tag);
self.refresh_activation_contract_digest();
Ok(self)
}
pub fn with_address_policy_digest(mut self, digest: impl Into<String>) -> anyhow::Result<Self> {
let digest = digest.into();
anyhow::ensure!(
digest.len() == 64 && digest.bytes().all(|byte| byte.is_ascii_hexdigit()),
"definition activation address policy digest is invalid"
);
self.address_policy_digest = Some(digest.to_ascii_lowercase());
self.refresh_activation_contract_digest();
Ok(self)
}
pub fn with_publisher_contract_digest(
mut self,
digest: impl Into<String>,
) -> anyhow::Result<Self> {
let digest = digest.into();
anyhow::ensure!(
digest.len() == 64 && digest.bytes().all(|byte| byte.is_ascii_hexdigit()),
"definition activation publisher contract digest is invalid"
);
self.publisher_contract_digest = Some(digest.to_ascii_lowercase());
self.refresh_activation_contract_digest();
Ok(self)
}
pub fn with_result_backend_digest(mut self, digest: impl Into<String>) -> anyhow::Result<Self> {
let digest = digest.into();
anyhow::ensure!(
self.transport == "kafka"
&& digest.len() == 64
&& digest.bytes().all(|byte| byte.is_ascii_hexdigit()),
"definition activation result backend digest is invalid"
);
self.result_backend_digest = Some(digest.to_ascii_lowercase());
self.refresh_activation_contract_digest();
Ok(self)
}
pub fn with_result_contract_digest(
mut self,
owner: impl Into<String>,
digest: impl Into<String>,
) -> anyhow::Result<Self> {
let owner = owner.into();
ServiceIdentity::new(&owner)?;
let digest = digest.into();
anyhow::ensure!(
digest.len() == 64 && digest.bytes().all(|byte| byte.is_ascii_hexdigit()),
"definition activation result contract digest is invalid"
);
self.result_contract_digests
.entry(owner)
.or_default()
.insert(digest.to_ascii_lowercase());
self.refresh_activation_contract_digest();
Ok(self)
}
fn refresh_activation_contract_digest(&mut self) {
let mut digest = Sha256::new();
update_snapshot_digest(&mut digest, b"nasaga-activation-contract");
update_snapshot_digest(&mut digest, self.transport.as_bytes());
match self.redis_key_tag.as_deref() {
Some(key_tag) => {
digest.update([1]);
update_snapshot_digest(&mut digest, key_tag.as_bytes());
}
None => digest.update([0]),
}
match self.address_policy_digest.as_deref() {
Some(policy_digest) => {
digest.update([1]);
update_snapshot_digest(&mut digest, policy_digest.as_bytes());
}
None => digest.update([0]),
}
match self.publisher_contract_digest.as_deref() {
Some(publisher_digest) => {
digest.update([1]);
update_snapshot_digest(&mut digest, publisher_digest.as_bytes());
}
None => digest.update([0]),
}
match self.result_backend_digest.as_deref() {
Some(result_backend_digest) => {
digest.update([1]);
update_snapshot_digest(&mut digest, result_backend_digest.as_bytes());
}
None => digest.update([0]),
}
digest.update((self.result_contract_digests.len() as u64).to_be_bytes());
for (owner, contracts) in &self.result_contract_digests {
update_snapshot_digest(&mut digest, owner.as_bytes());
digest.update((contracts.len() as u64).to_be_bytes());
for contract in contracts {
update_snapshot_digest(&mut digest, contract.as_bytes());
}
}
self.activation_contract_digest = hex::encode(digest.finalize());
}
pub fn orchestrator_service_identity(&self) -> &ServiceIdentity {
&self.orchestrator_service_identity
}
pub fn catalog_generation(&self) -> u64 {
self.catalog_generation
}
pub fn snapshot_digest(&self) -> &str {
&self.snapshot_digest
}
pub fn transport(&self) -> &str {
&self.transport
}
pub fn redis_key_tag(&self) -> Option<&str> {
self.redis_key_tag.as_deref()
}
pub fn result_backend_digest(&self) -> Option<&str> {
self.result_backend_digest.as_deref()
}
pub fn accepts_result_contract(&self, owner: &str, digest: &str) -> bool {
self.result_contract_digests
.get(owner)
.is_some_and(|contracts| contracts.contains(digest))
}
pub fn activation_contract_digest(&self) -> &str {
&self.activation_contract_digest
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CapabilityDescriptor {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub payload_contract: Option<nasaga_core::SagaPayloadContract>,
pub tenant: String,
pub owner: String,
pub replica_identity: String,
pub workflow: String,
pub definition_version: u32,
pub step: String,
pub compensation: DefinitionCompensation,
pub cancel_mode: DefinitionCancelMode,
pub allow_unknown: bool,
pub resolution_mode: Option<DefinitionResolutionMode>,
pub transport: String,
pub endpoint: String,
pub effective_saga_base_path: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub result_contract_digest: Option<String>,
pub route_generation: u64,
pub requested_lease_ms: u64,
}
impl CapabilityDescriptor {
pub fn digest(&self) -> String {
let mut canonical = self.clone();
canonical.requested_lease_ms = 0;
let bytes = serde_json::to_vec(&canonical)
.expect("serializing a capability descriptor cannot fail");
let mut hasher = Sha256::new();
hasher.update(bytes);
hex::encode(hasher.finalize())
}
pub fn route_contract_digest(&self) -> String {
let mut canonical = self.clone();
canonical.route_generation = 0;
canonical.requested_lease_ms = 0;
let bytes = serde_json::to_vec(&canonical)
.expect("serializing a capability descriptor cannot fail");
let mut hasher = Sha256::new();
hasher.update(bytes);
hex::encode(hasher.finalize())
}
pub fn matches_step(
&self,
tenant: &str,
definition: &WorkflowDefinition,
step: &StepDefinition,
) -> bool {
self.payload_contract.clone().unwrap_or_default() == *step.payload_contract()
&& self.tenant == tenant
&& self.workflow == definition.name().as_str()
&& self.definition_version == definition.version().get()
&& self.step == step.name().as_str()
&& self.owner == step.owner().as_str()
&& self.compensation == step.compensation().into()
&& self.cancel_mode == step.cancel_mode().into()
&& self.allow_unknown == step.resolution().allow_unknown()
&& self.resolution_mode == step.resolution().mode().map(Into::into)
}
}
pub fn validate_capability_route_contract(
descriptor: &CapabilityDescriptor,
) -> Result<(), DefinitionCatalogError> {
descriptor
.payload_contract
.clone()
.unwrap_or_default()
.validate()
.map_err(|_| DefinitionCatalogError::InvalidArgument)?;
let path_contract_is_valid = match descriptor.transport.as_str() {
"http" => descriptor
.effective_saga_base_path
.as_deref()
.is_some_and(valid_capability_http_base_path),
"grpc" | "kafka" | "redis-stream" => descriptor.effective_saga_base_path.is_none(),
_ => false,
};
if !path_contract_is_valid {
return Err(DefinitionCatalogError::InvalidArgument);
}
let result_contract_is_valid = match descriptor.transport.as_str() {
"http" | "grpc" | "kafka" | "redis-stream" => descriptor
.result_contract_digest
.as_deref()
.is_some_and(|digest| {
digest.len() == 64
&& digest == digest.to_ascii_lowercase()
&& digest.bytes().all(|byte| byte.is_ascii_hexdigit())
}),
_ => false,
};
if !result_contract_is_valid {
return Err(DefinitionCatalogError::InvalidArgument);
}
let endpoint_is_valid = match descriptor.transport.as_str() {
"http" => valid_capability_origin(&descriptor.endpoint, &["http", "https"]),
"grpc" => valid_capability_origin(&descriptor.endpoint, &["https"]),
"kafka" => {
!descriptor.endpoint.is_empty()
&& descriptor.endpoint.len() <= 249
&& descriptor.endpoint.trim() == descriptor.endpoint
&& !matches!(descriptor.endpoint.as_str(), "." | "..")
&& descriptor
.endpoint
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
}
"redis-stream" => validate_redis_stream_route(&descriptor.endpoint, None).is_ok(),
_ => false,
};
if endpoint_is_valid {
Ok(())
} else {
Err(DefinitionCatalogError::InvalidArgument)
}
}
fn valid_capability_origin(endpoint: &str, schemes: &[&str]) -> bool {
let raw_origin_is_canonical = endpoint
.split_once("://")
.is_some_and(|(scheme, authority)| {
schemes.contains(&scheme)
&& !authority.is_empty()
&& (authority.bytes().all(|byte| byte != b'/')
|| (authority.ends_with('/')
&& authority[..authority.len() - 1]
.bytes()
.all(|byte| byte != b'/')))
});
Url::parse(endpoint).is_ok_and(|url| {
endpoint.trim() == endpoint
&& raw_origin_is_canonical
&& schemes.contains(&url.scheme())
&& url.host_str().is_some_and(|host| !host.is_empty())
&& url.username().is_empty()
&& url.password().is_none()
&& url.query().is_none()
&& url.fragment().is_none()
&& matches!(url.path(), "" | "/")
})
}
pub fn validate_redis_stream_route(
stream: &str,
key_tag: Option<&str>,
) -> Result<(), DefinitionCatalogError> {
let valid = !stream.is_empty()
&& stream.len() <= 190
&& !stream.chars().any(char::is_control)
&& key_tag.is_none_or(|tag| {
valid_redis_key_tag(tag) && first_redis_hash_tag(stream) == Some(tag)
});
if valid {
Ok(())
} else {
Err(DefinitionCatalogError::InvalidArgument)
}
}
fn valid_redis_key_tag(tag: &str) -> bool {
!tag.is_empty()
&& tag.len() <= 64
&& !tag.contains(['{', '}'])
&& !tag.chars().any(char::is_control)
}
fn first_redis_hash_tag(name: &str) -> Option<&str> {
let open = name.find('{')?;
let close = name[open + 1..].find('}')?;
if close == 0 {
return None;
}
Some(&name[open + 1..open + 1 + close])
}
fn valid_capability_http_base_path(path: &str) -> bool {
path != "/"
&& path.starts_with('/')
&& !path.ends_with('/')
&& path.bytes().all(|byte| {
byte.is_ascii_alphanumeric() || matches!(byte, b'/' | b'-' | b'_' | b'.' | b'~')
})
&& !path.contains("//")
&& !path.contains('%')
&& !path.contains('?')
&& !path.contains('#')
&& !path.contains('*')
&& !path.contains('{')
&& !path.contains('}')
&& !path.split('/').any(|segment| matches!(segment, "." | ".."))
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CapabilityReceipt {
pub accepted_until_ms: i64,
pub capability_digest: String,
pub route_generation: u64,
pub catalog_generation: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RegisteredCapability {
pub descriptor: CapabilityDescriptor,
pub accepted_until_ms: i64,
pub capability_digest: String,
}
pub fn select_capability_route<'a>(
capabilities: &'a [RegisteredCapability],
transport: &str,
tenant: &str,
definition: &WorkflowDefinition,
step: &StepDefinition,
) -> Result<Option<&'a CapabilityDescriptor>, DefinitionCatalogError> {
let mut selected: Option<&CapabilityDescriptor> = None;
for descriptor in capabilities
.iter()
.map(|capability| &capability.descriptor)
.filter(|descriptor| {
descriptor.transport == transport && descriptor.matches_step(tenant, definition, step)
})
{
validate_capability_route_contract(descriptor)?;
if selected.is_some_and(|current| {
current.endpoint != descriptor.endpoint
|| current.effective_saga_base_path != descriptor.effective_saga_base_path
|| (transport == "kafka"
&& current.result_contract_digest != descriptor.result_contract_digest)
}) {
return Err(DefinitionCatalogError::FailedPrecondition);
}
selected.get_or_insert(descriptor);
}
Ok(selected)
}
pub fn select_capability_routes<'a>(
capabilities: &'a [RegisteredCapability],
transport: &str,
tenant: &str,
definition: &WorkflowDefinition,
step: &StepDefinition,
) -> Result<Vec<&'a CapabilityDescriptor>, DefinitionCatalogError> {
if !matches!(transport, "http" | "grpc") {
return select_capability_route(capabilities, transport, tenant, definition, step)
.map(|route| route.into_iter().collect());
}
let mut routes: Vec<&CapabilityDescriptor> = Vec::new();
for descriptor in capabilities
.iter()
.map(|capability| &capability.descriptor)
.filter(|descriptor| {
descriptor.transport == transport && descriptor.matches_step(tenant, definition, step)
})
{
validate_capability_route_contract(descriptor)?;
if routes.iter().any(|current| {
current.replica_identity == descriptor.replica_identity && *current != descriptor
}) {
return Err(DefinitionCatalogError::FailedPrecondition);
}
routes.push(descriptor);
}
routes.sort_by(|left, right| {
(&left.endpoint, &left.replica_identity).cmp(&(&right.endpoint, &right.replica_identity))
});
routes.dedup();
Ok(routes)
}
#[derive(Debug, Clone)]
pub struct DynamicCatalogSnapshot {
pub generation: u64,
pub snapshot_digest: String,
pub registry: DefinitionRegistry,
pub definition_records: Vec<DefinitionRecord>,
pub capabilities: Vec<RegisteredCapability>,
}
impl DynamicCatalogSnapshot {
pub fn new(
generation: u64,
registry: DefinitionRegistry,
capabilities: Vec<RegisteredCapability>,
) -> Self {
Self::with_definition_records(generation, registry, Vec::new(), capabilities)
}
pub fn with_definition_records(
generation: u64,
registry: DefinitionRegistry,
mut definition_records: Vec<DefinitionRecord>,
mut capabilities: Vec<RegisteredCapability>,
) -> Self {
definition_records.sort_by(|left, right| {
(
&left.artifact.tenant,
&left.artifact.workflow,
left.artifact.definition_version,
)
.cmp(&(
&right.artifact.tenant,
&right.artifact.workflow,
right.artifact.definition_version,
))
});
capabilities.sort_by(|left, right| {
(
&left.descriptor.tenant,
&left.descriptor.owner,
&left.descriptor.replica_identity,
&left.descriptor.workflow,
left.descriptor.definition_version,
&left.descriptor.step,
)
.cmp(&(
&right.descriptor.tenant,
&right.descriptor.owner,
&right.descriptor.replica_identity,
&right.descriptor.workflow,
right.descriptor.definition_version,
&right.descriptor.step,
))
});
let mut digest = Sha256::new();
digest.update(generation.to_be_bytes());
let mut runtime_definitions = registry
.definitions()
.map(|definition| {
(
definition.name().as_str(),
definition.version().get(),
definition.digest(),
)
})
.collect::<Vec<_>>();
runtime_definitions.sort();
for (workflow, version, definition_digest) in runtime_definitions {
update_snapshot_digest(&mut digest, workflow.as_bytes());
digest.update(version.to_be_bytes());
update_snapshot_digest(&mut digest, definition_digest.as_bytes());
}
for record in &definition_records {
update_snapshot_digest(&mut digest, record.artifact.tenant.as_bytes());
update_snapshot_digest(&mut digest, record.artifact.workflow_owner.as_bytes());
update_snapshot_digest(&mut digest, record.artifact.workflow.as_bytes());
digest.update(record.artifact.definition_version.to_be_bytes());
update_snapshot_digest(&mut digest, record.lifecycle.as_str().as_bytes());
update_snapshot_digest(&mut digest, record.artifact.seal.as_bytes());
digest.update(record.catalog_generation.to_be_bytes());
}
for capability in &capabilities {
update_snapshot_digest(&mut digest, capability.descriptor.tenant.as_bytes());
update_snapshot_digest(&mut digest, capability.descriptor.owner.as_bytes());
update_snapshot_digest(
&mut digest,
capability.descriptor.replica_identity.as_bytes(),
);
update_snapshot_digest(&mut digest, capability.descriptor.workflow.as_bytes());
digest.update(capability.descriptor.definition_version.to_be_bytes());
update_snapshot_digest(&mut digest, capability.descriptor.step.as_bytes());
update_snapshot_digest(&mut digest, capability.capability_digest.as_bytes());
}
Self {
generation,
snapshot_digest: hex::encode(digest.finalize()),
registry,
definition_records,
capabilities,
}
}
}
fn update_snapshot_digest(digest: &mut Sha256, value: &[u8]) {
digest.update((value.len() as u64).to_be_bytes());
digest.update(value);
}