1use schemars::JsonSchema;
2use serde::{Deserialize, Serialize};
3use serde_json::Value;
4use utoipa::ToSchema;
5
6use crate::{RestoreDecision, ServiceRestoreEvidence, extraction_input_digest};
7
8pub const DISASTER_RECOVERY_PLAN_PROTOCOL: &str = "lenso.disaster-recovery-plan.v1";
9pub const DISASTER_RECOVERY_EVIDENCE_PROTOCOL: &str = "lenso.disaster-recovery-evidence.v1";
10pub const DISASTER_RECOVERY_APPROVAL_BOUNDARY: &str = "single_region_disaster_cutover";
11pub const DISASTER_FAILBACK_APPROVAL_BOUNDARY: &str = "single_region_disaster_failback";
12
13#[derive(
14 Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema, ToSchema,
15)]
16#[serde(rename_all = "snake_case")]
17pub enum DisasterRecoveryPhase {
18 Cutover,
19 Failback,
20}
21
22#[derive(
23 Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema, ToSchema,
24)]
25#[serde(rename_all = "snake_case")]
26pub enum DisasterRecoveryDecision {
27 Ready,
28 Passed,
29 Blocked,
30}
31
32#[derive(
33 Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema, ToSchema,
34)]
35#[serde(rename_all = "snake_case")]
36pub enum DisasterRecoveryIssueCode {
37 RestoreEvidenceInvalid,
38 RegionTopologyInvalid,
39 PrimaryNotFenced,
40 PassiveNotReady,
41 ApprovalInvalid,
42 RecoveryBudgetExceeded,
43 IdentityOrContractMismatch,
44 FailbackPlanMissing,
45 PlanStale,
46 ActiveStateChanged,
47 ReconciliationIncomplete,
48 CleanupIncomplete,
49}
50
51#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
52#[serde(rename_all = "camelCase")]
53pub struct DisasterRecoveryIssue {
54 pub code: DisasterRecoveryIssueCode,
55 pub message: String,
56 pub remediation: String,
57 pub next_actions: Vec<String>,
58}
59
60#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
61#[serde(rename_all = "camelCase")]
62pub struct DisasterRecoveryPlanInput {
63 pub phase: DisasterRecoveryPhase,
64 pub service_id: String,
65 pub primary_region: String,
66 pub passive_region: String,
67 pub restore_evidence: ServiceRestoreEvidence,
68 pub expected_release_digest: String,
69 pub expected_config_revision_digest: String,
70 pub expected_contract_set_digest: String,
71 pub expected_active_state_digest: String,
72 pub authoritative_environment_count_before: u32,
73 pub planned_at_unix_ms: u64,
74 pub freshness_horizon_unix_ms: u64,
75 pub rpo_budget_ms: u64,
76 pub rto_budget_ms: u64,
77 pub primary_fenced: bool,
78 pub passive_fenced: bool,
79 pub passive_health_verified: bool,
80 pub passive_identity_verified: bool,
81 pub passive_contracts_verified: bool,
82 pub failback_steps: Vec<String>,
83}
84
85#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
86#[serde(rename_all = "camelCase")]
87pub struct DisasterRecoveryPlan {
88 pub protocol: String,
89 pub plan_id: String,
90 pub plan_digest: String,
91 #[serde(flatten)]
92 pub input: DisasterRecoveryPlanInput,
93 pub decision: DisasterRecoveryDecision,
94 pub issues: Vec<DisasterRecoveryIssue>,
95 pub approval_boundary: String,
96 pub effects: Vec<String>,
97}
98
99#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
100#[serde(rename_all = "camelCase")]
101pub struct DisasterRecoveryApproval {
102 pub plan_digest: String,
103 pub phase: DisasterRecoveryPhase,
104 pub approver: String,
105 pub reason: String,
106 pub approved_at_unix_ms: u64,
107}
108
109#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
110#[serde(rename_all = "camelCase")]
111pub struct DisasterRecoveryObservation {
112 pub plan_digest: String,
113 pub phase: DisasterRecoveryPhase,
114 pub observed_at_unix_ms: u64,
115 pub active_state_digest: String,
116 pub authoritative_environment_count: u32,
117 pub primary_fenced: bool,
118 pub passive_fenced: bool,
119 pub passive_became_authoritative: bool,
120 pub primary_became_authoritative: bool,
121 pub traffic_switched: bool,
122 pub observed_rpo_ms: u64,
123 pub observed_rto_ms: u64,
124 pub release_digest: String,
125 pub config_revision_digest: String,
126 pub contract_set_digest: String,
127 pub workload_identity_preserved: bool,
128 pub duplicate_business_effects: u64,
129 pub lost_committed_effects: u64,
130 pub requests_events_workflows_stories_verified: bool,
131 pub cleanup_complete: bool,
132 pub evidence_digest: String,
133}
134
135#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
136#[serde(rename_all = "camelCase")]
137pub struct DisasterRecoveryEvidence {
138 pub protocol: String,
139 pub evidence_id: String,
140 pub evidence_digest: String,
141 pub plan_id: String,
142 pub plan_digest: String,
143 pub phase: DisasterRecoveryPhase,
144 pub service_id: String,
145 pub primary_region: String,
146 pub passive_region: String,
147 pub observed_rpo_ms: u64,
148 pub observed_rto_ms: u64,
149 pub data_loss_bound_ms: u64,
150 pub replay_bound_count: u64,
151 pub cleanup_complete: bool,
152 pub authoritative_environment_count: u32,
153 pub decision: DisasterRecoveryDecision,
154 pub issues: Vec<DisasterRecoveryIssue>,
155 pub approval_boundary: String,
156 pub failback_steps: Vec<String>,
157}
158
159#[must_use]
160pub fn plan_disaster_recovery(input: DisasterRecoveryPlanInput) -> DisasterRecoveryPlan {
161 let mut issues = Vec::new();
162 if input.restore_evidence.decision != RestoreDecision::Passed
163 || input.restore_evidence.production_mutated
164 || input.restore_evidence.service_id != input.service_id
165 || !valid_digest(&input.restore_evidence.evidence_digest)
166 {
167 issues.push(issue(
168 DisasterRecoveryIssueCode::RestoreEvidenceInvalid,
169 "Disaster recovery lacks a verified passive restore for the exact Service.",
170 "Use immutable restore evidence from an isolated target Store.",
171 "Repeat backup and restore before planning cutover.",
172 ));
173 }
174 if !valid_digest(&input.expected_active_state_digest)
175 || input.authoritative_environment_count_before != 1
176 || input.planned_at_unix_ms == 0
177 || input.freshness_horizon_unix_ms < input.planned_at_unix_ms
178 {
179 issues.push(issue(
180 DisasterRecoveryIssueCode::PlanStale,
181 "The plan lacks a fresh exact active-state revision.",
182 "Bind the plan to the current authoritative state and freshness horizon.",
183 "Refresh active-state evidence before requesting approval.",
184 ));
185 }
186 if input.primary_region.trim().is_empty()
187 || input.passive_region.trim().is_empty()
188 || input.primary_region == input.passive_region
189 {
190 issues.push(issue(
191 DisasterRecoveryIssueCode::RegionTopologyInvalid,
192 "Active and passive regions are not distinct.",
193 "Declare one authoritative primary and one isolated passive region.",
194 "Correct the regional topology.",
195 ));
196 }
197 let source_fenced = match input.phase {
198 DisasterRecoveryPhase::Cutover => input.primary_fenced,
199 DisasterRecoveryPhase::Failback => input.passive_fenced,
200 };
201 if !source_fenced {
202 issues.push(issue(
203 DisasterRecoveryIssueCode::PrimaryNotFenced,
204 "The primary region can still accept authoritative writes.",
205 "Fence the primary before granting authority to the passive region.",
206 "Stop before cutover and verify the fencing observation.",
207 ));
208 }
209 if !input.passive_health_verified {
210 issues.push(issue(
211 DisasterRecoveryIssueCode::PassiveNotReady,
212 "The passive Workloads and restored Store are not ready.",
213 "Verify health while the passive remains non-authoritative.",
214 "Repair the passive region before requesting approval.",
215 ));
216 }
217 if !input.passive_identity_verified || !input.passive_contracts_verified {
218 issues.push(issue(
219 DisasterRecoveryIssueCode::IdentityOrContractMismatch,
220 "The passive region does not preserve Workload Identity or Contract identity.",
221 "Bind the passive region to the exact supported release and identities.",
222 "Correct the passive deployment before cutover.",
223 ));
224 }
225 if input.failback_steps.is_empty() {
226 issues.push(issue(
227 DisasterRecoveryIssueCode::FailbackPlanMissing,
228 "No stale-safe failback procedure is recorded.",
229 "Plan re-seeding, verification, fencing, approval, and traffic reversal.",
230 "Add the explicit failback steps before cutover.",
231 ));
232 }
233 let decision = if issues.is_empty() {
234 DisasterRecoveryDecision::Ready
235 } else {
236 DisasterRecoveryDecision::Blocked
237 };
238 let phase = input.phase;
239 let mut plan = DisasterRecoveryPlan {
240 protocol: DISASTER_RECOVERY_PLAN_PROTOCOL.to_owned(),
241 plan_id: String::new(),
242 plan_digest: String::new(),
243 input,
244 decision,
245 issues,
246 approval_boundary: approval_boundary(phase).to_owned(),
247 effects: match phase {
248 DisasterRecoveryPhase::Cutover => vec![
249 "fence primary".to_owned(),
250 "grant passive authority".to_owned(),
251 "switch regional traffic".to_owned(),
252 ],
253 DisasterRecoveryPhase::Failback => vec![
254 "fence passive".to_owned(),
255 "grant primary authority".to_owned(),
256 "switch regional traffic".to_owned(),
257 ],
258 },
259 };
260 plan.plan_digest = digest_without_plan_identity(&plan);
261 plan.plan_id = format!("disaster-recovery-plan:{}", &plan.plan_digest[7..23]);
262 plan
263}
264
265#[must_use]
266pub fn evaluate_disaster_recovery(
267 plan: &DisasterRecoveryPlan,
268 approval: &DisasterRecoveryApproval,
269 observation: DisasterRecoveryObservation,
270) -> DisasterRecoveryEvidence {
271 let mut issues = plan.issues.clone();
272 if plan.protocol != DISASTER_RECOVERY_PLAN_PROTOCOL
273 || !valid_digest(&plan.plan_digest)
274 || plan.plan_digest != digest_without_plan_identity(plan)
275 || plan.plan_id != format!("disaster-recovery-plan:{}", &plan.plan_digest[7..23])
276 {
277 issues.push(issue(
278 DisasterRecoveryIssueCode::RestoreEvidenceInvalid,
279 "Disaster recovery plan integrity is invalid.",
280 "Reject modified plans after review.",
281 "Regenerate the plan from current evidence.",
282 ));
283 }
284 if plan.decision != DisasterRecoveryDecision::Ready
285 || approval.plan_digest != plan.plan_digest
286 || approval.phase != plan.input.phase
287 || approval.approver.trim().is_empty()
288 || approval.reason.trim().is_empty()
289 || approval.approved_at_unix_ms == 0
290 {
291 issues.push(issue(
292 DisasterRecoveryIssueCode::ApprovalInvalid,
293 "Disaster cutover lacks explicit approval for the exact plan digest.",
294 "Obtain named human approval at the disaster-cutover boundary.",
295 "Stop without changing regional authority.",
296 ));
297 }
298 if observation.observed_at_unix_ms == 0
299 || observation.observed_at_unix_ms > plan.input.freshness_horizon_unix_ms
300 {
301 issues.push(issue(
302 DisasterRecoveryIssueCode::PlanStale,
303 "Cutover or failback observation is outside the plan freshness horizon.",
304 "Revalidate the exact plan immediately before authority mutation.",
305 "Stop and generate a new plan and approval.",
306 ));
307 }
308 if observation.active_state_digest != plan.input.expected_active_state_digest {
309 issues.push(issue(
310 DisasterRecoveryIssueCode::ActiveStateChanged,
311 "Authoritative state changed after the plan was reviewed.",
312 "Never apply a stale authority-transfer plan.",
313 "Refresh state, regenerate the plan, and obtain new approval.",
314 ));
315 }
316 let authority_transition_valid = match plan.input.phase {
317 DisasterRecoveryPhase::Cutover => {
318 observation.primary_fenced && observation.passive_became_authoritative
319 }
320 DisasterRecoveryPhase::Failback => {
321 observation.passive_fenced && observation.primary_became_authoritative
322 }
323 };
324 if observation.plan_digest != plan.plan_digest
325 || observation.phase != plan.input.phase
326 || !valid_digest(&observation.evidence_digest)
327 || !authority_transition_valid
328 || observation.authoritative_environment_count != 1
329 || !observation.traffic_switched
330 {
331 issues.push(issue(
332 DisasterRecoveryIssueCode::PrimaryNotFenced,
333 "Observed cutover does not prove fencing, passive authority, and traffic switch.",
334 "Collect one authoritative regional observation.",
335 "Repair or roll back the cutover before serving traffic.",
336 ));
337 }
338 if !observation.requests_events_workflows_stories_verified {
339 issues.push(issue(
340 DisasterRecoveryIssueCode::ReconciliationIncomplete,
341 "Requests, Events, Workflows, Inbox/Outbox, or Stories are not reconciled.",
342 "Verify every declared recovery outcome before completing the drill.",
343 "Keep the target isolated and finish reconciliation.",
344 ));
345 }
346 if !observation.cleanup_complete {
347 issues.push(issue(
348 DisasterRecoveryIssueCode::CleanupIncomplete,
349 "Disposable disaster-recovery resources remain active.",
350 "Clean both regional fixtures without changing production authority.",
351 "Complete deterministic cleanup.",
352 ));
353 }
354 if observation.observed_rpo_ms > plan.input.rpo_budget_ms
355 || observation.observed_rto_ms > plan.input.rto_budget_ms
356 {
357 issues.push(issue(
358 DisasterRecoveryIssueCode::RecoveryBudgetExceeded,
359 "Observed RPO or RTO exceeds the pinned environment budget.",
360 "Report the observation without converting it into a universal guarantee.",
361 "Improve the recovery path or revise the reviewed support envelope.",
362 ));
363 }
364 if observation.release_digest != plan.input.expected_release_digest
365 || observation.config_revision_digest != plan.input.expected_config_revision_digest
366 || observation.contract_set_digest != plan.input.expected_contract_set_digest
367 || !observation.workload_identity_preserved
368 || observation.duplicate_business_effects > 0
369 || observation.lost_committed_effects > 0
370 {
371 issues.push(issue(
372 DisasterRecoveryIssueCode::IdentityOrContractMismatch,
373 "Recovered authority changed identity or lost or duplicated committed work.",
374 "Preserve release, configuration, Contract, Workload Identity, Inbox, and Outbox boundaries.",
375 "Fail closed and restore the last verified authority state.",
376 ));
377 }
378 let decision = if issues.is_empty() {
379 DisasterRecoveryDecision::Passed
380 } else {
381 DisasterRecoveryDecision::Blocked
382 };
383 let mut evidence = DisasterRecoveryEvidence {
384 protocol: DISASTER_RECOVERY_EVIDENCE_PROTOCOL.to_owned(),
385 evidence_id: String::new(),
386 evidence_digest: String::new(),
387 plan_id: plan.plan_id.clone(),
388 plan_digest: plan.plan_digest.clone(),
389 phase: plan.input.phase,
390 service_id: plan.input.service_id.clone(),
391 primary_region: plan.input.primary_region.clone(),
392 passive_region: plan.input.passive_region.clone(),
393 observed_rpo_ms: observation.observed_rpo_ms,
394 observed_rto_ms: observation.observed_rto_ms,
395 data_loss_bound_ms: observation.observed_rpo_ms,
396 replay_bound_count: observation.duplicate_business_effects,
397 cleanup_complete: observation.cleanup_complete,
398 authoritative_environment_count: observation.authoritative_environment_count,
399 decision,
400 issues,
401 approval_boundary: approval_boundary(plan.input.phase).to_owned(),
402 failback_steps: plan.input.failback_steps.clone(),
403 };
404 evidence.evidence_digest = digest_without_evidence_identity(&evidence);
405 evidence.evidence_id = format!("disaster-recovery:{}", &evidence.evidence_digest[7..23]);
406 evidence
407}
408
409const fn approval_boundary(phase: DisasterRecoveryPhase) -> &'static str {
410 match phase {
411 DisasterRecoveryPhase::Cutover => DISASTER_RECOVERY_APPROVAL_BOUNDARY,
412 DisasterRecoveryPhase::Failback => DISASTER_FAILBACK_APPROVAL_BOUNDARY,
413 }
414}
415
416#[must_use]
417pub fn disaster_recovery_evidence_schema() -> Value {
418 let mut schema = serde_json::to_value(schemars::schema_for!(DisasterRecoveryEvidence))
419 .expect("disaster recovery schema serializes");
420 schema["$id"] = Value::String(
421 "https://contracts.lenso.local/ga/lenso.disaster-recovery-evidence.v1.schema.json"
422 .to_owned(),
423 );
424 schema
425}
426
427fn issue(
428 code: DisasterRecoveryIssueCode,
429 message: impl Into<String>,
430 remediation: impl Into<String>,
431 next_action: impl Into<String>,
432) -> DisasterRecoveryIssue {
433 DisasterRecoveryIssue {
434 code,
435 message: message.into(),
436 remediation: remediation.into(),
437 next_actions: vec![next_action.into()],
438 }
439}
440
441fn valid_digest(value: &str) -> bool {
442 value.strip_prefix("sha256:").is_some_and(|digest| {
443 digest.len() == 64 && digest.bytes().all(|byte| byte.is_ascii_hexdigit())
444 })
445}
446
447fn digest_json(value: &impl Serialize) -> String {
448 extraction_input_digest(&serde_json::to_vec(value).expect("DR evidence serializes"))
449}
450
451fn digest_without_plan_identity(plan: &DisasterRecoveryPlan) -> String {
452 let mut canonical = plan.clone();
453 canonical.plan_id.clear();
454 canonical.plan_digest.clear();
455 digest_json(&canonical)
456}
457
458fn digest_without_evidence_identity(evidence: &DisasterRecoveryEvidence) -> String {
459 let mut canonical = evidence.clone();
460 canonical.evidence_id.clear();
461 canonical.evidence_digest.clear();
462 digest_json(&canonical)
463}
464
465#[must_use]
466pub fn disaster_recovery_evidence_integrity_is_valid(evidence: &DisasterRecoveryEvidence) -> bool {
467 valid_digest(&evidence.evidence_digest)
468 && evidence.evidence_digest == digest_without_evidence_identity(evidence)
469 && evidence.evidence_id == format!("disaster-recovery:{}", &evidence.evidence_digest[7..23])
470}