nasaga-runtime-core 2.0.0

Backend-neutral NASA Saga state machine and shared runtime state
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
//! 动态 Definition Catalog 与参与方能力目录使用的公开数据合同。
//!
//! 完整流程顺序只来自 [`DefinitionArtifact`]。参与方能力只能证明某个 owner 当前能够执行
//! 其中一个步骤,不能反向拼装流程。

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;

/// capability 审计对象键的最大字符数:tenant、四个结构性身份、u32 十进制版本和五个分隔符。
pub const CATALOG_OBJECT_KEY_MAX_LEN: usize =
    OPAQUE_IDENTIFIER_MAX_BYTES + STRUCTURED_IDENTIFIER_MAX_BYTES * 4 + 10 + 5;

/// 业务作用:表示 Catalog 中不可变 definition 的生命周期。
#[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,
}

/// 业务作用:封闭 Definition Catalog 可由协议层安全公开的确定性拒绝类别。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum DefinitionCatalogError {
    /// 请求字段不满足 Catalog 的封闭格式或范围合同。
    InvalidArgument,
    /// 认证主体不是目标 definition 或 capability 的持有者。
    PermissionDenied,
    /// 目标 definition 不存在。
    NotFound,
    /// 生命周期、集群确认或 capability 等业务前置条件尚未成立。
    FailedPrecondition,
    /// 相同 tenant/workflow/version 已绑定另一份不可变摘要。
    DigestConflict,
    /// 调用方预期的 definition seal 已不再是当前持久事实。
    PreconditionFailed,
    /// 相同操作身份已绑定另一组动作参数。
    OperationConflict,
}

impl DefinitionCatalogError {
    /// 业务作用:从持久事务上下文包裹的错误链提取 Catalog 确定性拒绝。
    ///
    /// 参数说明:`error` 是 Catalog 写入口返回的完整错误链。
    ///
    /// 返回:命中封闭拒绝时返回对应类别;其它失败返回 `None`。
    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 {
    /// 业务作用:输出不携带 definition 正文或租户信息的稳定 Catalog 拒绝摘要。
    ///
    /// 参数说明:`formatter` 是标准格式化输出目标。
    ///
    /// 返回:摘要写入成功时返回 `Ok`;格式化失败返回对应错误。
    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,
    /// 步骤是不可撤销 pivot。
    NonCompensable,
}

impl From<DefinitionCompensation> for Compensation {
    /// 业务作用:把控制面补偿声明转换为状态机强类型。
    ///
    /// 参数说明:`value` 是公开补偿枚举。
    ///
    /// 返回:对应的核心补偿类型。
    fn from(value: DefinitionCompensation) -> Self {
        match value {
            DefinitionCompensation::Compensable => Self::Compensable,
            DefinitionCompensation::NonCompensable => Self::NonCompensable,
        }
    }
}

impl From<Compensation> for DefinitionCompensation {
    /// 业务作用:把状态机补偿类型投影为可持久化字段。
    ///
    /// 参数说明:`value` 是已校验步骤的补偿能力。
    ///
    /// 返回:稳定的控制面补偿枚举。
    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 {
    /// 本地 gate 与业务写竞争同一事务。
    LocalFenceable,
    /// 通过外部幂等取消命令建立屏障。
    ExternallyCancellable,
    /// 无法取消,只能查询或对账。
    ResolveOnly,
}

impl From<DefinitionCancelMode> for CancelMode {
    /// 业务作用:把控制面取消声明转换为状态机强类型。
    ///
    /// 参数说明:`value` 是公开取消枚举。
    ///
    /// 返回:对应的核心取消形态。
    fn from(value: DefinitionCancelMode) -> Self {
        match value {
            DefinitionCancelMode::LocalFenceable => Self::LocalFenceable,
            DefinitionCancelMode::ExternallyCancellable => Self::ExternallyCancellable,
            DefinitionCancelMode::ResolveOnly => Self::ResolveOnly,
        }
    }
}

impl From<CancelMode> for DefinitionCancelMode {
    /// 业务作用:把状态机取消形态投影为可持久化字段。
    ///
    /// 参数说明:`value` 是已校验步骤的取消形态。
    ///
    /// 返回:稳定的控制面取消枚举。
    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 {
    /// 业务作用:把控制面解决模式转换为状态机强类型。
    ///
    /// 参数说明:`value` 是公开解决模式。
    ///
    /// 返回:对应的核心解决模式。
    fn from(value: DefinitionResolutionMode) -> Self {
        match value {
            DefinitionResolutionMode::Callback => Self::Callback,
            DefinitionResolutionMode::Poll => Self::Poll,
            DefinitionResolutionMode::Manual => Self::Manual,
        }
    }
}

impl From<ResolutionMode> for DefinitionResolutionMode {
    /// 业务作用:把状态机解决模式投影为可持久化字段。
    ///
    /// 参数说明:`value` 是已校验步骤的解决通道。
    ///
    /// 返回:稳定的控制面解决枚举。
    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`,表示本地事务结果总是可知。
    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 {
    /// 业务作用:把控制面超时策略转换为状态机强类型。
    ///
    /// 参数说明:`value` 是公开超时策略。
    ///
    /// 返回:对应的核心策略。
    fn from(value: DefinitionTimeoutPolicy) -> Self {
        match value {
            DefinitionTimeoutPolicy::AcceptLateSuccess => Self::AcceptLateSuccess,
            DefinitionTimeoutPolicy::CompensateLateSuccess => Self::CompensateLateSuccess,
        }
    }
}

impl From<TimeoutPolicy> for DefinitionTimeoutPolicy {
    /// 业务作用:把状态机超时策略投影为可持久化字段。
    ///
    /// 参数说明:`value` 是已校验步骤的迟到结果策略。
    ///
    /// 返回:稳定的控制面超时策略。
    fn from(value: TimeoutPolicy) -> Self {
        match value {
            TimeoutPolicy::AcceptLateSuccess => Self::AcceptLateSuccess,
            TimeoutPolicy::CompensateLateSuccess => Self::CompensateLateSuccess,
        }
    }
}

/// 业务作用:完整描述 definition 中一个有序步骤的业务和失败语义。
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DefinitionStepArtifact {
    /// 步骤正文合同;缺省为无 schema 的 JSON,非默认值参与 definition seal。
    #[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,
}

/// 业务作用:承载 workflow owner 发布的完整、有序且带 seal 的流程合同。
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DefinitionArtifact {
    /// 定义适用的租户域。
    pub tenant: String,
    /// 获授权发布完整流程的逻辑主体。
    pub workflow_owner: String,
    /// 稳定 workflow 名称。
    pub workflow: String,
    /// 从一开始递增的不可变定义版本。
    pub definition_version: u32,
    /// 按正向顺序排列的完整步骤集合。
    pub steps: Vec<DefinitionStepArtifact>,
    /// 发布者根据完整 definition 计算的 canonical 摘要。
    pub seal: String,
}

impl DefinitionArtifact {
    /// 业务作用:转换为状态机 definition,并复验 seal 覆盖完整业务语义。
    ///
    /// 参数说明: 无。
    ///
    /// 返回:字段、流程不变量与 seal 全部成立时返回 definition;否则拒绝进入 Catalog。
    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)
    }

    /// 业务作用:从已校验 definition 生成可发布 artifact,确保 seal 与实际步骤合同同源。
    ///
    /// 参数说明:`tenant` 是租户域,`workflow_owner` 是发布主体,`definition` 是完整流程。
    ///
    /// 返回:字段顺序稳定且 seal 已计算的 artifact。
    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(),
        }
    }
}

/// 业务作用:返回持久 Catalog 中 definition 的内容与当前 generation。
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DefinitionRecord {
    /// 完整流程 artifact。
    pub artifact: DefinitionArtifact,
    /// 当前生命周期。
    pub lifecycle: DefinitionLifecycle,
    /// 最近一次生命周期变更形成的 Catalog generation。
    pub catalog_generation: u64,
    /// definition 首次持久化的数据库时间。
    pub published_at_ms: i64,
    /// definition 首次进入 active 的数据库时间;候选尚未激活时为空。
    pub activated_at_ms: Option<i64>,
}

/// 业务作用:冻结一次 definition 激活必须满足的集群快照与数据面路由合同。
#[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,
}

/// 业务作用:把人工 definition 生命周期动作的 CAS、幂等身份与审计原因冻结为一个持久裁决输入。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DefinitionLifecycleOperation {
    expected_sha256: String,
    operation_id: String,
    reason: String,
}

impl DefinitionLifecycleOperation {
    /// 业务作用:校验并构造可进入 Catalog 事务的生命周期操作合同。
    ///
    /// 参数说明:`expected_sha256` 是调用方观察到的 seal,`operation_id` 是主体范围内的幂等键,
    /// `reason` 是必须持久保存的人工动作原因。
    ///
    /// 返回:字段格式和长度均安全时返回合同;空值、非法摘要或超长文本返回错误。
    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,
        })
    }

    /// 业务作用:为 validated 自动激活构造绑定 definition seal、身份和 Catalog generation 的定长幂等操作。
    ///
    /// 参数说明:`artifact` 是待激活的完整 definition,`catalog_generation` 是已确认快照,`reason` 是持久审计原因。
    ///
    /// 返回:返回长度不受租户或流程名影响的合法操作;seal 或原因不符合生命周期合同时返回错误。
    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,
        )
    }

    /// 业务作用:返回必须在持有 definition 行锁时复验的 seal。
    ///
    /// 参数说明: 无。
    ///
    /// 返回:规范化的小写 SHA-256。
    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 {
    /// 业务作用:构造只能由同代 Ready 副本和指定数据面共同满足的激活门禁。
    ///
    /// 参数说明:`orchestrator_service_identity` 定位逻辑控制面,`catalog_generation` 与
    /// `snapshot_digest` 固定副本已确认快照,`transport` 固定本次部署的数据面协议。
    ///
    /// 返回:摘要和协议属于封闭集合时返回门禁;无效输入拒绝进入数据库激活事务。
    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)
    }

    /// 业务作用:把 Redis Cluster 的实际同槽配置绑定进 definition 激活门禁,确保数据库裁决
    /// 与稍后构造 publisher 使用同一份 route 上下文。
    ///
    /// 参数说明:`key_tag` 是运行配置要求 capability stream 首个生效 hash tag 匹配的值。
    ///
    /// 返回:仅在 transport 为 Redis Streams 且 tag 满足名称合同时返回门禁;其它组合返回错误。
    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)
    }

    /// 业务作用:把当前 transport 的受信地址政策绑定进副本确认与 definition 激活门禁。
    ///
    /// 参数说明:`digest` 是调用方对 transport 专属 canonical allowlist 计算的 SHA-256。
    ///
    /// 返回:摘要合法时返回更新后的门禁;格式无效时拒绝形成可持久确认的运行时合同。
    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)
    }

    /// 业务作用:把最终 publisher 使用的路由、凭据、client 与超时合同绑定进副本确认,
    /// 防止同一 Catalog 快照被运行参数不同的副本同时发布为权威。
    ///
    /// 参数说明:`digest` 是调用方对已解析 publisher 输入计算的 canonical SHA-256。
    ///
    /// 返回:摘要格式合法时返回更新后的门禁;格式无效时拒绝形成持久确认。
    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)
    }

    /// 业务作用:把 Kafka result 返回路径的后端身份绑定进激活事务,确保参与方能力与
    /// Orchestrator consumer 使用同一套 broker 命名边界。
    ///
    /// 参数说明:`digest` 是规范化 broker 地址集合与 result topic 前缀形成的 SHA-256。
    ///
    /// 返回:仅在 transport 为 Kafka 且摘要格式合法时返回门禁;其它组合拒绝激活。
    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)
    }

    /// 业务作用:把一个 participant owner 的 result 发送凭据与 Orchestrator 接收凭据绑定进激活事务。
    ///
    /// 参数说明:`owner` 是 definition step 的服务身份,`digest` 是协议专属凭据合同的 SHA-256。
    ///
    /// 返回:身份与摘要合法时返回门禁;非法输入拒绝形成副本确认合同。
    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)
    }

    /// 业务作用:重算副本必须共同确认的运行时路由合同摘要。
    ///
    /// 参数说明: 无。
    ///
    /// 返回:无;transport、Redis tag、地址政策与 publisher 合同按带类型边界的固定顺序写入摘要。
    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());
    }

    /// 业务作用:返回有权确认本次激活快照的逻辑 Orchestrator 身份。
    ///
    /// 参数说明: 无。
    ///
    /// 返回:构造门禁时已校验的长期服务身份。
    pub fn orchestrator_service_identity(&self) -> &ServiceIdentity {
        &self.orchestrator_service_identity
    }

    /// 业务作用:返回待激活 definition 所依赖的精确 Catalog generation。
    ///
    /// 参数说明: 无。
    ///
    /// 返回:副本确认和数据库状态必须共同命中的代际。
    pub fn catalog_generation(&self) -> u64 {
        self.catalog_generation
    }

    /// 业务作用:返回全部有效 Ready 副本必须一致确认的完整快照摘要。
    ///
    /// 参数说明: 无。
    ///
    /// 返回:六十四位十六进制摘要。
    pub fn snapshot_digest(&self) -> &str {
        &self.snapshot_digest
    }

    /// 业务作用:返回 definition 每个步骤在激活时必须具备的受管数据面协议。
    ///
    /// 参数说明: 无。
    ///
    /// 返回:HTTP、gRPC、Kafka 或 Redis Streams 的稳定协议名。
    pub fn transport(&self) -> &str {
        &self.transport
    }

    /// 业务作用:返回 Redis Streams 激活时必须由 capability route 满足的 Cluster 同槽 tag。
    ///
    /// 参数说明: 无。
    ///
    /// 返回:Redis 激活门禁绑定的 tag;其它 transport 或尚未补齐上下文时为空。
    pub fn redis_key_tag(&self) -> Option<&str> {
        self.redis_key_tag.as_deref()
    }

    /// 业务作用:返回 Kafka candidate 必须匹配的 result broker 与 topic 命名合同摘要。
    ///
    /// 参数说明: 无。
    ///
    /// 返回:Kafka 激活门禁已绑定的摘要;其它 transport 返回空。
    pub fn result_backend_digest(&self) -> Option<&str> {
        self.result_backend_digest.as_deref()
    }

    /// 业务作用:判断 capability 声明的 result 凭据合同是否属于该 owner 的受信接收集合。
    ///
    /// 参数说明:`owner` 是步骤服务身份,`digest` 是 participant 发布的协议专属合同摘要。
    ///
    /// 返回:门禁中存在完全一致的 owner 与摘要时为真。
    pub fn accepts_result_contract(&self, owner: &str, digest: &str) -> bool {
        self.result_contract_digests
            .get(owner)
            .is_some_and(|contracts| contracts.contains(digest))
    }

    /// 业务作用:返回全部 Ready 副本必须一致确认的数据面与地址政策摘要。
    ///
    /// 参数说明: 无。
    ///
    /// 返回:绑定 transport、Redis tag、地址政策和最终 publisher 输入的 canonical SHA-256。
    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 {
    /// 本 handler 接受的正文合同,必须与 definition 中对应步骤一致。
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub payload_contract: Option<nasaga_core::SagaPayloadContract>,
    /// 能力获准服务的租户,必须与 definition key 和授权范围一致。
    pub tenant: String,
    /// 能力所属逻辑服务身份,必须与认证主体一致。
    pub owner: String,
    /// 当前副本的稳定身份,用于租约续租。
    pub replica_identity: String,
    /// 支持的 workflow。
    pub workflow: String,
    /// 支持的 definition 版本。
    pub definition_version: u32,
    /// 支持的步骤名称。
    pub step: String,
    /// 步骤补偿能力。
    pub compensation: DefinitionCompensation,
    /// 步骤取消形态。
    pub cancel_mode: DefinitionCancelMode,
    /// 是否托管未知结果解决入口。
    pub allow_unknown: bool,
    /// 托管的解决模式;禁止未知时为空。
    pub resolution_mode: Option<DefinitionResolutionMode>,
    /// command transport 的稳定名称。
    pub transport: String,
    /// 当前实例 origin 或 broker route,不包含凭据。
    pub endpoint: String,
    /// HTTP 模式下实际生效的 Saga 基础路径。
    pub effective_saga_base_path: Option<String>,
    /// result producer 与 Orchestrator 接收端共同持有的协议专属凭据或后端合同摘要。
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub result_contract_digest: Option<String>,
    /// endpoint 与路径绑定的单调 route generation;持久值由 Catalog 在主键行锁内分配。
    pub route_generation: u64,
    /// 客户端建议的租约时长;服务端执行有界裁剪。
    pub requested_lease_ms: u64,
}

impl CapabilityDescriptor {
    /// 业务作用:计算不受续租建议影响的 capability 内容摘要,供幂等续租与 route 变更检测。
    ///
    /// 参数说明: 无。
    ///
    /// 返回:覆盖 owner、步骤合同、transport、endpoint 与 route generation 的小写十六进制摘要。
    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())
    }

    /// 业务作用:计算不含租约建议和已分配代际的路由合同摘要,供 Catalog 判断真实内容是否变化。
    ///
    /// 参数说明: 无。
    ///
    /// 返回:覆盖身份、步骤语义、transport、endpoint 与有效路径的稳定小写十六进制摘要。
    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())
    }

    /// 业务作用:复验 capability 与 definition 中被引用步骤的不可变语义一致。
    ///
    /// 参数说明:`tenant` 是候选流程的授权租户,`definition` 是候选流程,`step` 是其中一个步骤。
    ///
    /// 返回:owner、步骤身份、版本与能力完全一致时为真。
    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)
    }
}

/// 业务作用:按 command transport 复验 capability 的完整可发布 route 合同,确保目录准入、
/// candidate 激活与实际 publisher 对同一 descriptor 得出一致结论。
///
/// 参数说明:`descriptor` 携带 transport、endpoint、HTTP 专属有效路径与 result 凭据合同摘要。
///
/// 返回:HTTP origin 与路径、gRPC HTTPS origin、Kafka topic 或 Redis stream 名称完整合法时成功;
/// transport 专属字段缺失、越界或出现在其它 transport 时返回参数错误。
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)
    }
}

/// 业务作用:判断 capability 网络 endpoint 是否为 publisher 与地址政策都能安全解释的纯 origin。
///
/// 参数说明:`endpoint` 是目录输入,`schemes` 是当前 transport 允许的协议集合。
///
/// 返回:URI 含合法 host、允许协议且不携带凭据、路径、query 或 fragment 时为真。
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(), "" | "/")
    })
}

/// 业务作用:按 Redis Cluster 实际选 slot 的规则校验 stream 名与运行配置的同槽合同。
///
/// 参数说明:`stream` 是 capability 或 publisher 使用的完整 key,`key_tag` 是可选的受信配置值。
///
/// 返回:名称有界,且提供 tag 时首个生效 hash tag 精确匹配则成功;否则返回参数错误。
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)
    }
}

/// 业务作用:校验运行配置中的 Redis Cluster hash tag 可安全嵌入 stream key。
///
/// 参数说明:`tag` 是不含花括号的配置值。
///
/// 返回:非空、有界、无控制字符和花括号时为真。
fn valid_redis_key_tag(tag: &str) -> bool {
    !tag.is_empty()
        && tag.len() <= 64
        && !tag.contains(['{', '}'])
        && !tag.chars().any(char::is_control)
}

/// 业务作用:按 Redis 服务端规则解析 key 首对花括号形成的实际 hash tag,保持 activation 与
/// publisher 的 Cluster slot 裁决一致。
///
/// 参数说明:`name` 是完整 Redis key。
///
/// 返回:首对花括号之间非空时返回 tag;空 tag、未闭合或不存在时返回空。
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])
}

/// 业务作用:判断 HTTP capability 的有效 Saga path 是否能无歧义地追加固定操作名并参与签名。
///
/// 参数说明:`path` 是已经包含应用 context 的目录路径。
///
/// 返回:非根、无尾斜杠、编码、模板或特殊路径段的 canonical path 为真。
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, "." | ".."))
}

/// 业务作用:返回服务端接受的 capability 租约与幂等摘要。
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CapabilityReceipt {
    /// 服务端数据库时钟计算的租约截止时间。
    pub accepted_until_ms: i64,
    /// 不含租约截止时间的 capability 内容摘要。
    pub capability_digest: String,
    /// Catalog 在 capability 主键行锁内确认的单调 route generation。
    pub route_generation: u64,
    /// 本次写入所处 Catalog generation。
    pub catalog_generation: u64,
}

/// 业务作用:保存能力目录中一个仍在租约内的逐实例 route。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RegisteredCapability {
    /// 参与方发布的完整能力描述。
    pub descriptor: CapabilityDescriptor,
    /// 服务端数据库时钟确定的租约截止时间。
    pub accepted_until_ms: i64,
    /// 持久化内容摘要。
    pub capability_digest: String,
}

/// 业务作用:从有效能力快照中按完整步骤身份选择唯一数据面 route,避免同租户同名 endpoint
/// 的无关能力替代目标 descriptor 参与地址政策裁决。
///
/// 参数说明:
/// - `capabilities`: 当前 Catalog generation 内仍在租约期的完整能力集合。
/// - `transport`: 当前数据面的稳定协议名。
/// - `tenant`: definition 所属租户。
/// - `definition`: 当前调度实例固定的完整 definition。
/// - `step`: definition 中待选择 route 的完整步骤合同。
///
/// 返回:没有匹配能力时返回 `Ok(None)`;所有匹配副本声明同一合法 endpoint 与有效路径,且 Kafka
/// result 后端摘要一致时返回其中一个完整 descriptor;route 非法或出现不同目标时返回错误。
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)
}

/// 业务作用:取得同一步骤的完整逐实例直接路由集合,允许 HTTP/gRPC 副本使用独立 endpoint。
/// 参数说明:`capabilities` 是仍在租期内的同代快照;`transport`、`tenant`、`definition`、`step` 共同限定步骤合同。
/// 返回:合法 HTTP/gRPC 路由按 endpoint 与副本身份排序;消息代理仍要求唯一目标,非法或同副本冲突时拒绝。
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)
}

/// 业务作用:把同一数据库 generation 的 definition 与有效 capability 形成不可拆分加载结果。
#[derive(Debug, Clone)]
pub struct DynamicCatalogSnapshot {
    /// 本次读取确认的共享 generation。
    pub generation: u64,
    /// definition 与 route 内容形成的确定性摘要,供多副本确认同一代快照。
    pub snapshot_digest: String,
    /// active 与仍被旧实例引用的 deprecated definition 快照。
    pub registry: DefinitionRegistry,
    /// 当前 generation 中候选与可运行 definition 的完整持久化记录。
    pub definition_records: Vec<DefinitionRecord>,
    /// 读取时仍在服务端租约内的参与方逐实例能力。
    pub capabilities: Vec<RegisteredCapability>,
}

impl DynamicCatalogSnapshot {
    /// 业务作用:把已复验的 definition 与 capability 组合为带确定性内容摘要的代际快照。
    ///
    /// 参数说明:`generation` 是共享代际,`registry` 是运行定义,`capabilities` 是有效路由。
    ///
    /// 返回:返回可供多副本确认的不可拆分快照。
    pub fn new(
        generation: u64,
        registry: DefinitionRegistry,
        capabilities: Vec<RegisteredCapability>,
    ) -> Self {
        Self::with_definition_records(generation, registry, Vec::new(), capabilities)
    }

    /// 业务作用:把 candidate、active、deprecated definition 与有效 capability 共同绑定进快照摘要。
    ///
    /// 参数说明:`generation` 是共享代际,`registry` 只承载可运行定义,`definition_records` 还包含
    /// 尚未激活的候选定义,`capabilities` 是读取时有效的逐实例路由。
    ///
    /// 返回:摘要覆盖租户、owner、生命周期、seal 和路由且字段边界无歧义的不可拆分快照。
    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());
        // 可运行 registry 也必须进入摘要,否则静态快照或租户投影漂移时仍可伪装成同代内容。
        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,
        }
    }
}

/// 业务作用:以长度前缀把可变字段加入 Catalog 摘要,避免相邻字段拼接产生边界歧义。
///
/// 参数说明:`digest` 是当前 SHA-256 状态,`value` 是一个完整领域字段的原始字节。
///
/// 返回:无;字段长度和内容按固定顺序写入摘要状态。
fn update_snapshot_digest(digest: &mut Sha256, value: &[u8]) {
    digest.update((value.len() as u64).to_be_bytes());
    digest.update(value);
}