Skip to main content

nasaga_runtime_core/
catalog.rs

1//! 动态 Definition Catalog 与参与方能力目录使用的公开数据合同。
2//!
3//! 完整流程顺序只来自 [`DefinitionArtifact`]。参与方能力只能证明某个 owner 当前能够执行
4//! 其中一个步骤,不能反向拼装流程。
5
6use 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
20/// capability 审计对象键的最大字符数:tenant、四个结构性身份、u32 十进制版本和五个分隔符。
21pub const CATALOG_OBJECT_KEY_MAX_LEN: usize =
22    OPAQUE_IDENTIFIER_MAX_BYTES + STRUCTURED_IDENTIFIER_MAX_BYTES * 4 + 10 + 5;
23
24/// 业务作用:表示 Catalog 中不可变 definition 的生命周期。
25#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
26#[serde(rename_all = "snake_case")]
27pub enum DefinitionLifecycle {
28    /// 已持久化但尚不能创建实例。
29    Candidate,
30    /// 已通过控制面门禁,可以创建新实例。
31    Active,
32    /// 禁止创建新实例,但继续服务已有实例。
33    Deprecated,
34    /// 已确认不存在运行期引用,不再装载。
35    Retired,
36}
37
38impl DefinitionLifecycle {
39    /// 业务作用:返回数据库和协议共同使用的稳定名称。
40    ///
41    /// 参数说明: 无。
42    ///
43    /// 返回:生命周期的稳定文本。
44    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/// 业务作用:表达定义发布的幂等裁决,调用方无需解析错误文本。
55#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
56#[serde(rename_all = "snake_case")]
57pub enum DefinitionPublishDisposition {
58    /// 首次写入候选定义。
59    Published,
60    /// 相同键与摘要已存在,持久内容未变化。
61    Duplicate,
62}
63
64/// 业务作用:封闭 Definition Catalog 可由协议层安全公开的确定性拒绝类别。
65#[derive(Debug, Clone, Copy, PartialEq, Eq)]
66#[non_exhaustive]
67pub enum DefinitionCatalogError {
68    /// 请求字段不满足 Catalog 的封闭格式或范围合同。
69    InvalidArgument,
70    /// 认证主体不是目标 definition 或 capability 的持有者。
71    PermissionDenied,
72    /// 目标 definition 不存在。
73    NotFound,
74    /// 生命周期、集群确认或 capability 等业务前置条件尚未成立。
75    FailedPrecondition,
76    /// 相同 tenant/workflow/version 已绑定另一份不可变摘要。
77    DigestConflict,
78    /// 调用方预期的 definition seal 已不再是当前持久事实。
79    PreconditionFailed,
80    /// 相同操作身份已绑定另一组动作参数。
81    OperationConflict,
82}
83
84impl DefinitionCatalogError {
85    /// 业务作用:从持久事务上下文包裹的错误链提取 Catalog 确定性拒绝。
86    ///
87    /// 参数说明:
88    /// - `error`:是 Catalog 写入口返回的完整错误链。
89    ///
90    /// 返回:命中封闭拒绝时返回对应类别;其它失败返回 `None`。
91    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    /// 业务作用:输出不携带 definition 正文或租户信息的稳定 Catalog 拒绝摘要。
100    ///
101    /// 参数说明:
102    /// - `formatter`:是标准格式化输出目标。
103    ///
104    /// 返回:摘要写入成功时返回 `Ok`;格式化失败返回对应错误。
105    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/// 业务作用:声明步骤的补偿能力。
123#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
124#[serde(rename_all = "kebab-case")]
125pub enum DefinitionCompensation {
126    /// 步骤成功后可进入逆序补偿计划。
127    Compensable,
128    /// 步骤是不可撤销 pivot。
129    NonCompensable,
130}
131
132impl From<DefinitionCompensation> for Compensation {
133    /// 业务作用:把控制面补偿声明转换为状态机强类型。
134    ///
135    /// 参数说明:
136    /// - `value`:是公开补偿枚举。
137    ///
138    /// 返回:对应的核心补偿类型。
139    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    /// 业务作用:把状态机补偿类型投影为可持久化字段。
149    ///
150    /// 参数说明:
151    /// - `value`:是已校验步骤的补偿能力。
152    ///
153    /// 返回:稳定的控制面补偿枚举。
154    fn from(value: Compensation) -> Self {
155        match value {
156            Compensation::Compensable => Self::Compensable,
157            Compensation::NonCompensable => Self::NonCompensable,
158        }
159    }
160}
161
162/// 业务作用:声明步骤的取消屏障形态。
163#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
164#[serde(rename_all = "kebab-case")]
165pub enum DefinitionCancelMode {
166    /// 本地 gate 与业务写竞争同一事务。
167    LocalFenceable,
168    /// 通过外部幂等取消命令建立屏障。
169    ExternallyCancellable,
170    /// 无法取消,只能查询或对账。
171    ResolveOnly,
172}
173
174impl From<DefinitionCancelMode> for CancelMode {
175    /// 业务作用:把控制面取消声明转换为状态机强类型。
176    ///
177    /// 参数说明:
178    /// - `value`:是公开取消枚举。
179    ///
180    /// 返回:对应的核心取消形态。
181    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    /// 业务作用:把状态机取消形态投影为可持久化字段。
192    ///
193    /// 参数说明:
194    /// - `value`:是已校验步骤的取消形态。
195    ///
196    /// 返回:稳定的控制面取消枚举。
197    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/// 业务作用:声明未知结果的解决通道。
207#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
208#[serde(rename_all = "snake_case")]
209pub enum DefinitionResolutionMode {
210    /// 外部系统主动回调确定结果。
211    Callback,
212    /// 参与方按稳定效果身份主动查询。
213    Poll,
214    /// 交由受审计人工裁决。
215    Manual,
216}
217
218impl From<DefinitionResolutionMode> for ResolutionMode {
219    /// 业务作用:把控制面解决模式转换为状态机强类型。
220    ///
221    /// 参数说明:
222    /// - `value`:是公开解决模式。
223    ///
224    /// 返回:对应的核心解决模式。
225    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    /// 业务作用:把状态机解决模式投影为可持久化字段。
236    ///
237    /// 参数说明:
238    /// - `value`:是已校验步骤的解决通道。
239    ///
240    /// 返回:稳定的控制面解决枚举。
241    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/// 业务作用:表达禁止未知结果或为未知结果配置有界解决通道。
251#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
252#[serde(untagged)]
253pub enum DefinitionResolution {
254    /// 必须是文本 `forbidden`,表示本地事务结果总是可知。
255    Forbidden(String),
256    /// 允许未知结果,并声明解决通道和总预算。
257    Allowed {
258        /// 解决通道。
259        mode: DefinitionResolutionMode,
260        /// 从首次未知结果开始计算的总预算毫秒数。
261        budget_ms: u64,
262    },
263}
264
265/// 业务作用:声明步骤超时后的迟到成功裁决。
266#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
267#[serde(rename_all = "kebab-case")]
268pub enum DefinitionTimeoutPolicy {
269    /// 接受迟到成功并继续正向流程。
270    AcceptLateSuccess,
271    /// 把迟到成功纳入冻结补偿计划。
272    CompensateLateSuccess,
273}
274
275impl From<DefinitionTimeoutPolicy> for TimeoutPolicy {
276    /// 业务作用:把控制面超时策略转换为状态机强类型。
277    ///
278    /// 参数说明:
279    /// - `value`:是公开超时策略。
280    ///
281    /// 返回:对应的核心策略。
282    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    /// 业务作用:把状态机超时策略投影为可持久化字段。
292    ///
293    /// 参数说明:
294    /// - `value`:是已校验步骤的迟到结果策略。
295    ///
296    /// 返回:稳定的控制面超时策略。
297    fn from(value: TimeoutPolicy) -> Self {
298        match value {
299            TimeoutPolicy::AcceptLateSuccess => Self::AcceptLateSuccess,
300            TimeoutPolicy::CompensateLateSuccess => Self::CompensateLateSuccess,
301        }
302    }
303}
304
305/// 业务作用:完整描述 definition 中一个有序步骤的业务和失败语义。
306#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
307#[serde(deny_unknown_fields)]
308pub struct DefinitionStepArtifact {
309    /// 步骤正文合同;缺省为无 schema 的 JSON,非默认值参与 definition seal。
310    #[serde(default, skip_serializing_if = "Option::is_none")]
311    pub payload_contract: Option<nasaga_core::SagaPayloadContract>,
312    /// 稳定步骤名称。
313    pub name: String,
314    /// 唯一有权作证结果的逻辑服务身份。
315    pub owner: String,
316    /// 补偿能力。
317    pub compensation: DefinitionCompensation,
318    /// 取消屏障形态。
319    pub cancel_mode: DefinitionCancelMode,
320    /// 未知结果解决声明。
321    pub resolution: DefinitionResolution,
322    /// 正向执行期限毫秒数。
323    pub timeout_ms: u64,
324    /// 迟到成功策略。
325    pub timeout_policy: DefinitionTimeoutPolicy,
326}
327
328/// 业务作用:承载 workflow owner 发布的完整、有序且带 seal 的流程合同。
329#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
330#[serde(deny_unknown_fields)]
331pub struct DefinitionArtifact {
332    /// 定义适用的租户域。
333    pub tenant: String,
334    /// 获授权发布完整流程的逻辑主体。
335    pub workflow_owner: String,
336    /// 稳定 workflow 名称。
337    pub workflow: String,
338    /// 从一开始递增的不可变定义版本。
339    pub definition_version: u32,
340    /// 按正向顺序排列的完整步骤集合。
341    pub steps: Vec<DefinitionStepArtifact>,
342    /// 发布者根据完整 definition 计算的 canonical 摘要。
343    pub seal: String,
344}
345
346impl DefinitionArtifact {
347    /// 业务作用:转换为状态机 definition,并复验 seal 覆盖完整业务语义。
348    ///
349    /// 参数说明: 无。
350    ///
351    /// 返回:字段、流程不变量与 seal 全部成立时返回 definition;否则拒绝进入 Catalog。
352    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    /// 业务作用:从已校验 definition 生成可发布 artifact,确保 seal 与实际步骤合同同源。
394    ///
395    /// 参数说明:
396    /// - `tenant`:是租户域
397    /// - `workflow_owner`:是发布主体
398    /// - `definition`:是完整流程。
399    ///
400    /// 返回:字段顺序稳定且 seal 已计算的 artifact。
401    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/// 业务作用:返回持久 Catalog 中 definition 的内容与当前 generation。
441#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
442#[serde(deny_unknown_fields)]
443pub struct DefinitionRecord {
444    /// 完整流程 artifact。
445    pub artifact: DefinitionArtifact,
446    /// 当前生命周期。
447    pub lifecycle: DefinitionLifecycle,
448    /// 最近一次生命周期变更形成的 Catalog generation。
449    pub catalog_generation: u64,
450    /// definition 首次持久化的数据库时间。
451    pub published_at_ms: i64,
452    /// definition 首次进入 active 的数据库时间;候选尚未激活时为空。
453    pub activated_at_ms: Option<i64>,
454}
455
456/// 业务作用:冻结一次 definition 激活必须满足的集群快照与数据面路由合同。
457#[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/// 业务作用:把人工 definition 生命周期动作的 CAS、幂等身份与审计原因冻结为一个持久裁决输入。
472#[derive(Debug, Clone, PartialEq, Eq)]
473pub struct DefinitionLifecycleOperation {
474    expected_sha256: String,
475    operation_id: String,
476    reason: String,
477}
478
479impl DefinitionLifecycleOperation {
480    /// 业务作用:校验并冻结一次定义生命周期人工动作的摘要及幂等身份。
481    ///
482    /// 参数说明:
483    /// - `expected_sha256`:调用方观察到的定义 seal,必须是 64 位十六进制文本。
484    /// - `operation_id`:主体范围内的稳定操作幂等键,须非空、无首尾空白或控制字符且不超过 190 字节。
485    /// - `reason`:需持久记录的人工动作原因,须非空、无首尾空白或控制字符且不超过 512 字节。
486    ///
487    /// 返回:合法时返回 seal 已转小写的操作合同,否则返回校验错误;不读取数据库或执行动作。
488    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    /// 业务作用:为 validated 自动激活构造绑定 definition seal、身份和 Catalog generation 的定长幂等操作。
523    ///
524    /// 参数说明:
525    /// - `artifact`:是待激活的完整 definition
526    /// - `catalog_generation`:是已确认快照
527    /// - `reason`:是持久审计原因。
528    ///
529    /// 返回:返回长度不受租户或流程名影响的合法操作;seal 或原因不符合生命周期合同时返回错误。
530    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    /// 业务作用:返回必须在持有 definition 行锁时复验的 seal。
557    ///
558    /// 参数说明: 无。
559    ///
560    /// 返回:规范化的小写 SHA-256。
561    pub fn expected_sha256(&self) -> &str {
562        &self.expected_sha256
563    }
564
565    /// 业务作用:返回人工生命周期动作的幂等身份。
566    ///
567    /// 参数说明: 无。
568    ///
569    /// 返回:已通过长度与非空校验的稳定身份。
570    pub fn operation_id(&self) -> &str {
571        &self.operation_id
572    }
573
574    /// 业务作用:返回与生命周期动作不可分持久化的审计原因。
575    ///
576    /// 参数说明: 无。
577    ///
578    /// 返回:已通过长度与非空校验的原因文本。
579    pub fn reason(&self) -> &str {
580        &self.reason
581    }
582}
583
584impl DefinitionActivationGate {
585    /// 业务作用:构造关联控制服务、目录快照和数据面协议的定义激活门禁。
586    ///
587    /// 参数说明:
588    /// - `orchestrator_service_identity`:需要确认同代快照的逻辑编排服务身份。
589    /// - `catalog_generation`:期望副本确认的共享目录代际,构造时原样保存。
590    /// - `snapshot_digest`:需要副本确认的 64 位十六进制快照摘要。
591    /// - `transport`:只允许 http、grpc、kafka 或 redis-stream 的数据面协议。
592    ///
593    /// 返回:摘要及协议有效时返回已计算激活合同摘要的门禁;非法时返回错误。可选数据面约束初始为空,不在此核实副本 Ready 或数据库事实。
594    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    /// 业务作用:把 Redis Cluster 的实际同槽配置绑定进 definition 激活门禁,确保数据库裁决
631    /// 与稍后构造 publisher 使用同一份 route 上下文。
632    ///
633    /// 参数说明:
634    /// - `key_tag`:是运行配置要求 capability stream 首个生效 hash tag 匹配的值。
635    ///
636    /// 返回:仅在 transport 为 Redis Streams 且 tag 满足名称合同时返回门禁;其它组合返回错误。
637    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    /// 业务作用:把当前 transport 的受信地址政策绑定进副本确认与 definition 激活门禁。
649    ///
650    /// 参数说明:
651    /// - `digest`:是调用方对 transport 专属 canonical allowlist 计算的 SHA-256。
652    ///
653    /// 返回:摘要合法时返回更新后的门禁;格式无效时拒绝形成可持久确认的运行时合同。
654    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    /// 业务作用:把最终 publisher 使用的路由、凭据、client 与超时合同绑定进副本确认,
666    /// 防止同一 Catalog 快照被运行参数不同的副本同时发布为权威。
667    ///
668    /// 参数说明:
669    /// - `digest`:是调用方对已解析 publisher 输入计算的 canonical SHA-256。
670    ///
671    /// 返回:摘要格式合法时返回更新后的门禁;格式无效时拒绝形成持久确认。
672    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    /// 业务作用:把 Kafka result 返回路径的后端身份绑定进激活事务,确保参与方能力与
687    /// Orchestrator consumer 使用同一套 broker 命名边界。
688    ///
689    /// 参数说明:
690    /// - `digest`:是规范化 broker 地址集合与 result topic 前缀形成的 SHA-256。
691    ///
692    /// 返回:仅在 transport 为 Kafka 且摘要格式合法时返回门禁;其它组合拒绝激活。
693    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    /// 业务作用:把一个 participant owner 的 result 发送凭据与 Orchestrator 接收凭据绑定进激活事务。
707    ///
708    /// 参数说明:
709    /// - `owner`:是 definition step 的服务身份
710    /// - `digest`:是协议专属凭据合同的 SHA-256。
711    ///
712    /// 返回:身份与摘要合法时返回门禁;非法输入拒绝形成副本确认合同。
713    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    /// 业务作用:重算副本必须共同确认的运行时路由合同摘要。
734    ///
735    /// 参数说明: 无。
736    ///
737    /// 返回:无;transport、Redis tag、地址政策与 publisher 合同按带类型边界的固定顺序写入摘要。
738    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    /// 业务作用:返回有权确认本次激活快照的逻辑 Orchestrator 身份。
782    ///
783    /// 参数说明: 无。
784    ///
785    /// 返回:构造门禁时已校验的长期服务身份。
786    pub fn orchestrator_service_identity(&self) -> &ServiceIdentity {
787        &self.orchestrator_service_identity
788    }
789
790    /// 业务作用:返回待激活 definition 所依赖的精确 Catalog generation。
791    ///
792    /// 参数说明: 无。
793    ///
794    /// 返回:副本确认和数据库状态必须共同命中的代际。
795    pub fn catalog_generation(&self) -> u64 {
796        self.catalog_generation
797    }
798
799    /// 业务作用:返回全部有效 Ready 副本必须一致确认的完整快照摘要。
800    ///
801    /// 参数说明: 无。
802    ///
803    /// 返回:六十四位十六进制摘要。
804    pub fn snapshot_digest(&self) -> &str {
805        &self.snapshot_digest
806    }
807
808    /// 业务作用:返回 definition 每个步骤在激活时必须具备的受管数据面协议。
809    ///
810    /// 参数说明: 无。
811    ///
812    /// 返回:HTTP、gRPC、Kafka 或 Redis Streams 的稳定协议名。
813    pub fn transport(&self) -> &str {
814        &self.transport
815    }
816
817    /// 业务作用:返回 Redis Streams 激活时必须由 capability route 满足的 Cluster 同槽 tag。
818    ///
819    /// 参数说明: 无。
820    ///
821    /// 返回:Redis 激活门禁绑定的 tag;其它 transport 或尚未补齐上下文时为空。
822    pub fn redis_key_tag(&self) -> Option<&str> {
823        self.redis_key_tag.as_deref()
824    }
825
826    /// 业务作用:返回 Kafka candidate 必须匹配的 result broker 与 topic 命名合同摘要。
827    ///
828    /// 参数说明: 无。
829    ///
830    /// 返回:Kafka 激活门禁已绑定的摘要;其它 transport 返回空。
831    pub fn result_backend_digest(&self) -> Option<&str> {
832        self.result_backend_digest.as_deref()
833    }
834
835    /// 业务作用:判断 capability 声明的 result 凭据合同是否属于该 owner 的受信接收集合。
836    ///
837    /// 参数说明:
838    /// - `owner`:是步骤服务身份
839    /// - `digest`:是 participant 发布的协议专属合同摘要。
840    ///
841    /// 返回:门禁中存在完全一致的 owner 与摘要时为真。
842    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    /// 业务作用:返回全部 Ready 副本必须一致确认的数据面与地址政策摘要。
849    ///
850    /// 参数说明: 无。
851    ///
852    /// 返回:绑定 transport、Redis tag、地址政策和最终 publisher 输入的 canonical SHA-256。
853    pub fn activation_contract_digest(&self) -> &str {
854        &self.activation_contract_digest
855    }
856}
857
858/// 业务作用:声明参与方当前可执行步骤的协议和逐实例地址事实。
859#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
860#[serde(deny_unknown_fields)]
861pub struct CapabilityDescriptor {
862    /// 本 handler 接受的正文合同,必须与 definition 中对应步骤一致。
863    #[serde(default, skip_serializing_if = "Option::is_none")]
864    pub payload_contract: Option<nasaga_core::SagaPayloadContract>,
865    /// 能力获准服务的租户,必须与 definition key 和授权范围一致。
866    pub tenant: String,
867    /// 能力所属逻辑服务身份,必须与认证主体一致。
868    pub owner: String,
869    /// 当前副本的稳定身份,用于租约续租。
870    pub replica_identity: String,
871    /// 支持的 workflow。
872    pub workflow: String,
873    /// 支持的 definition 版本。
874    pub definition_version: u32,
875    /// 支持的步骤名称。
876    pub step: String,
877    /// 步骤补偿能力。
878    pub compensation: DefinitionCompensation,
879    /// 步骤取消形态。
880    pub cancel_mode: DefinitionCancelMode,
881    /// 是否托管未知结果解决入口。
882    pub allow_unknown: bool,
883    /// 托管的解决模式;禁止未知时为空。
884    pub resolution_mode: Option<DefinitionResolutionMode>,
885    /// command transport 的稳定名称。
886    pub transport: String,
887    /// 当前实例 origin 或 broker route,不包含凭据。
888    pub endpoint: String,
889    /// HTTP 模式下实际生效的 Saga 基础路径。
890    pub effective_saga_base_path: Option<String>,
891    /// result producer 与 Orchestrator 接收端共同持有的协议专属凭据或后端合同摘要。
892    #[serde(default, skip_serializing_if = "Option::is_none")]
893    pub result_contract_digest: Option<String>,
894    /// endpoint 与路径绑定的单调 route generation;持久值由 Catalog 在主键行锁内分配。
895    pub route_generation: u64,
896    /// 客户端建议的租约时长;服务端执行有界裁剪。
897    pub requested_lease_ms: u64,
898}
899
900impl CapabilityDescriptor {
901    /// 业务作用:计算不受续租建议影响的 capability 内容摘要,供幂等续租与 route 变更检测。
902    ///
903    /// 参数说明: 无。
904    ///
905    /// 返回:覆盖 owner、步骤合同、transport、endpoint 与 route generation 的小写十六进制摘要。
906    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    /// 业务作用:计算不含租约建议和已分配代际的路由合同摘要,供 Catalog 判断真实内容是否变化。
917    ///
918    /// 参数说明: 无。
919    ///
920    /// 返回:覆盖身份、步骤语义、transport、endpoint 与有效路径的稳定小写十六进制摘要。
921    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    /// 业务作用:复验 capability 与 definition 中被引用步骤的不可变语义一致。
933    ///
934    /// 参数说明:
935    /// - `tenant`:是候选流程的授权租户
936    /// - `definition`:是候选流程
937    /// - `step`:是其中一个步骤。
938    ///
939    /// 返回:owner、步骤身份、版本与能力完全一致时为真。
940    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
959/// 业务作用:按 command transport 复验 capability 的完整可发布 route 合同,确保目录准入、
960/// candidate 激活与实际 publisher 对同一 descriptor 得出一致结论。
961///
962/// 参数说明:
963/// - `descriptor`:携带 transport、endpoint、HTTP 专属有效路径与 result 凭据合同摘要。
964///
965/// 返回:HTTP origin 与路径、gRPC HTTPS origin、Kafka topic 或 Redis stream 名称完整合法时成功;
966/// transport 专属字段缺失、越界或出现在其它 transport 时返回参数错误。
967pub 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
1025/// 业务作用:判断 capability 网络 endpoint 是否为 publisher 与地址政策都能安全解释的纯 origin。
1026///
1027/// 参数说明:
1028/// - `endpoint`:是目录输入
1029/// - `schemes`:是当前 transport 允许的协议集合。
1030///
1031/// 返回:URI 含合法 host、允许协议且不携带凭据、路径、query 或 fragment 时为真。
1032fn 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
1057/// 业务作用:按 Redis Cluster 实际选 slot 的规则校验 stream 名与运行配置的同槽合同。
1058///
1059/// 参数说明:
1060/// - `stream`:是 capability 或 publisher 使用的完整 key
1061/// - `key_tag`:是可选的受信配置值。
1062///
1063/// 返回:名称有界,且提供 tag 时首个生效 hash tag 精确匹配则成功;否则返回参数错误。
1064pub 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
1081/// 业务作用:校验运行配置中的 Redis Cluster hash tag 可安全嵌入 stream key。
1082///
1083/// 参数说明:
1084/// - `tag`:是不含花括号的配置值。
1085///
1086/// 返回:非空、有界、无控制字符和花括号时为真。
1087fn 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
1094/// 业务作用:按 Redis 服务端规则解析 key 首对花括号形成的实际 hash tag,保持 activation 与
1095/// publisher 的 Cluster slot 裁决一致。
1096///
1097/// 参数说明:
1098/// - `name`:是完整 Redis key。
1099///
1100/// 返回:首对花括号之间非空时返回 tag;空 tag、未闭合或不存在时返回空。
1101fn 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
1110/// 业务作用:判断 HTTP capability 的有效 Saga path 是否能无歧义地追加固定操作名并参与签名。
1111///
1112/// 参数说明:
1113/// - `path`:是已经包含应用 context 的目录路径。
1114///
1115/// 返回:非根、无尾斜杠、编码、模板或特殊路径段的 canonical path 为真。
1116fn 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/// 业务作用:返回服务端接受的 capability 租约与幂等摘要。
1134#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1135#[serde(deny_unknown_fields)]
1136pub struct CapabilityReceipt {
1137    /// 服务端数据库时钟计算的租约截止时间。
1138    pub accepted_until_ms: i64,
1139    /// 不含租约截止时间的 capability 内容摘要。
1140    pub capability_digest: String,
1141    /// Catalog 在 capability 主键行锁内确认的单调 route generation。
1142    pub route_generation: u64,
1143    /// 本次写入所处 Catalog generation。
1144    pub catalog_generation: u64,
1145}
1146
1147/// 业务作用:保存能力目录中一个仍在租约内的逐实例 route。
1148#[derive(Debug, Clone, PartialEq, Eq)]
1149pub struct RegisteredCapability {
1150    /// 参与方发布的完整能力描述。
1151    pub descriptor: CapabilityDescriptor,
1152    /// 服务端数据库时钟确定的租约截止时间。
1153    pub accepted_until_ms: i64,
1154    /// 持久化内容摘要。
1155    pub capability_digest: String,
1156}
1157
1158/// 业务作用:从有效能力快照中按完整步骤身份选择唯一数据面 route,避免同租户同名 endpoint
1159/// 的无关能力替代目标 descriptor 参与地址政策裁决。
1160///
1161/// 参数说明:
1162/// - `capabilities`: 当前 Catalog generation 内仍在租约期的完整能力集合。
1163/// - `transport`: 当前数据面的稳定协议名。
1164/// - `tenant`: definition 所属租户。
1165/// - `definition`: 当前调度实例固定的完整 definition。
1166/// - `step`: definition 中待选择 route 的完整步骤合同。
1167///
1168/// 返回:没有匹配能力时返回 `Ok(None)`;所有匹配副本声明同一合法 endpoint 与有效路径,且 Kafka
1169/// result 后端摘要一致时返回其中一个完整 descriptor;route 非法或出现不同目标时返回错误。
1170pub 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
1199/// 业务作用:从给定能力快照选择与步骤合同匹配的直接路由。
1200///
1201/// 参数说明:
1202/// - `capabilities`:调用方已经确认租约与代际有效的能力记录,本函数不复验租期。
1203/// - `transport`:本次需要的 HTTP、gRPC 或消息代理协议。
1204/// - `tenant`:必须与能力声明匹配的受信租户。
1205/// - `definition`:提供工作流名、定义版本及摘要的冻结定义。
1206/// - `step`:本次要执行的冻结步骤合同。
1207///
1208/// 返回:HTTP/gRPC 返回按端点及副本排序去重的路由引用;同副本声明互斥合同或路由非法时拒绝。消息代理委托单路由选择,允许空结果,不发布路由或执行网络。
1209pub 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        // 同一副本不能同时自报互斥地址或结果合同,否则不得随机挑选其中一份权威。
1230        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/// 业务作用:把同一数据库 generation 的 definition 与有效 capability 形成不可拆分加载结果。
1245#[derive(Debug, Clone)]
1246pub struct DynamicCatalogSnapshot {
1247    /// 本次读取确认的共享 generation。
1248    pub generation: u64,
1249    /// definition 与 route 内容形成的确定性摘要,供多副本确认同一代快照。
1250    pub snapshot_digest: String,
1251    /// active 与仍被旧实例引用的 deprecated definition 快照。
1252    pub registry: DefinitionRegistry,
1253    /// 当前 generation 中候选与可运行 definition 的完整持久化记录。
1254    pub definition_records: Vec<DefinitionRecord>,
1255    /// 读取时仍在服务端租约内的参与方逐实例能力。
1256    pub capabilities: Vec<RegisteredCapability>,
1257}
1258
1259impl DynamicCatalogSnapshot {
1260    /// 业务作用:把已复验的 definition 与 capability 组合为带确定性内容摘要的代际快照。
1261    ///
1262    /// 参数说明:
1263    /// - `generation`:是共享代际
1264    /// - `registry`:是运行定义
1265    /// - `capabilities`:是有效路由。
1266    ///
1267    /// 返回:返回可供多副本确认的不可拆分快照。
1268    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    /// 业务作用:把运行注册表、定义生命周期记录和逐副本能力组合为带内容摘要的目录快照。
1277    ///
1278    /// 参数说明:
1279    /// - `generation`:调用方为整份快照指定的共享目录代际。
1280    /// - `registry`:当前可运行的工作流定义注册表。
1281    /// - `definition_records`:包含候选、活跃及弃用状态的完整定义记录。
1282    /// - `capabilities`:调用方已验证可用的逐实例能力记录。
1283    ///
1284    /// 返回:返回保留排序后记录及注册表的拥有式快照,摘要覆盖目录代际、可运行定义、定义记录和能力摘要;构造时不验证租户一致性、生命周期转换或租约有效性。
1285    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        // 可运行 registry 也必须进入摘要,否则静态快照或租户投影漂移时仍可伪装成同代内容。
1324        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
1371/// 业务作用:以长度前缀把可变字段加入 Catalog 摘要,避免相邻字段拼接产生边界歧义。
1372///
1373/// 参数说明:
1374/// - `digest`:是当前 SHA-256 状态
1375/// - `value`:是一个完整领域字段的原始字节。
1376///
1377/// 返回:无;字段长度和内容按固定顺序写入摘要状态。
1378fn update_snapshot_digest(digest: &mut Sha256, value: &[u8]) {
1379    digest.update((value.len() as u64).to_be_bytes());
1380    digest.update(value);
1381}