1use chrono::{DateTime, Utc};
19use serde::{Deserialize, Serialize};
20use serde_json::Value;
21use thiserror::Error;
22
23use crate::EnvId;
24use crate::audit::{Actor, AuditDecision, AuditEvent};
25use crate::integrity::{IntegrityError, StateIntegrity};
26use crate::version::SchemaVersion;
27
28#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
32#[serde(transparent)]
33pub struct StateEtag(pub String);
34
35impl StateEtag {
36 pub fn of<T: Serialize>(value: &T) -> Result<Self, IntegrityError> {
38 Ok(Self(StateIntegrity::sha256_of(value)?.digest))
39 }
40
41 pub fn from_integrity(integrity: &StateIntegrity) -> Self {
43 Self(integrity.digest.clone())
44 }
45
46 pub fn header_value(&self) -> String {
48 format!("\"{}\"", self.0)
49 }
50}
51
52#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
56pub struct Precondition {
57 #[serde(skip_serializing_if = "Option::is_none")]
58 pub if_match: Option<StateEtag>,
59 #[serde(skip_serializing_if = "Option::is_none")]
60 pub expected_generation: Option<u64>,
61}
62
63impl Precondition {
64 pub fn matching(etag: StateEtag, generation: u64) -> Self {
66 Self {
67 if_match: Some(etag),
68 expected_generation: Some(generation),
69 }
70 }
71
72 pub fn is_conditional(&self) -> bool {
75 self.if_match.is_some() || self.expected_generation.is_some()
76 }
77
78 pub fn check(
87 &self,
88 current_etag: &StateEtag,
89 current_generation: u64,
90 ) -> Result<(), PreconditionError> {
91 if !self.is_conditional() {
92 return Err(PreconditionError::Required);
93 }
94 let etag_ok = self.if_match.as_ref().is_none_or(|e| e == current_etag);
95 let gen_ok = self
96 .expected_generation
97 .is_none_or(|g| g == current_generation);
98 if etag_ok && gen_ok {
99 Ok(())
100 } else {
101 Err(PreconditionError::Conflict(ConcurrencyConflict {
102 expected_etag: self.if_match.as_ref().map(|e| e.0.clone()),
103 actual_etag: current_etag.0.clone(),
104 expected_generation: self.expected_generation,
105 actual_generation: current_generation,
106 }))
107 }
108 }
109}
110
111#[derive(Debug, Clone, PartialEq, Eq, Error)]
113pub enum PreconditionError {
114 #[error("a conditional write must pin If-Match and/or expected generation")]
117 Required,
118 #[error("precondition failed (stale generation/etag)")]
120 Conflict(ConcurrencyConflict),
121}
122
123#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
125pub struct ConcurrencyConflict {
126 #[serde(skip_serializing_if = "Option::is_none")]
127 pub expected_etag: Option<String>,
128 pub actual_etag: String,
129 #[serde(skip_serializing_if = "Option::is_none")]
130 pub expected_generation: Option<u64>,
131 pub actual_generation: u64,
132}
133
134#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
137#[serde(try_from = "String", into = "String")]
138pub struct IdempotencyKey(String);
139
140impl IdempotencyKey {
141 pub fn new(key: impl Into<String>) -> Result<Self, RemoteContractError> {
142 let key = key.into();
143 if key.trim().is_empty() {
144 return Err(RemoteContractError::EmptyIdempotencyKey);
145 }
146 Ok(Self(key))
147 }
148
149 pub fn as_str(&self) -> &str {
150 &self.0
151 }
152}
153
154impl TryFrom<String> for IdempotencyKey {
155 type Error = RemoteContractError;
156 fn try_from(value: String) -> Result<Self, Self::Error> {
157 Self::new(value)
158 }
159}
160
161impl From<IdempotencyKey> for String {
162 fn from(key: IdempotencyKey) -> Self {
163 key.0
164 }
165}
166
167#[derive(Debug, Clone, Serialize, Deserialize)]
176pub struct IdempotencyRecord {
177 pub key: IdempotencyKey,
178 pub request_fingerprint: String,
181 pub response: MutationResponse,
183 pub stored_at: DateTime<Utc>,
184}
185
186impl IdempotencyRecord {
187 pub fn fingerprint<T: Serialize>(request: &T) -> Result<String, IntegrityError> {
189 Ok(StateIntegrity::sha256_of(request)?.digest)
190 }
191
192 pub fn match_request(&self, incoming_fingerprint: &str) -> IdempotencyReplay<'_> {
196 if self.request_fingerprint == incoming_fingerprint {
197 IdempotencyReplay::Replay(&self.response)
198 } else {
199 IdempotencyReplay::Conflict {
200 reason: "idempotency key reused with a different request body".to_string(),
201 }
202 }
203 }
204}
205
206#[derive(Debug)]
208pub enum IdempotencyReplay<'a> {
209 Replay(&'a MutationResponse),
212 Conflict { reason: String },
215}
216
217#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
221#[serde(tag = "idempotency", rename_all = "kebab-case")]
222pub enum IdempotencyOutcome {
223 Applied,
225 Replayed,
228}
229
230#[derive(Debug, Clone, Serialize, Deserialize)]
233pub struct RbacRequest {
234 pub actor: Actor,
235 pub env_id: EnvId,
236 pub noun: String,
237 pub verb: String,
238 pub target: Value,
239}
240
241#[derive(Debug, Clone, Serialize, Deserialize)]
245pub struct MutationResponse {
246 pub etag: StateEtag,
247 pub generation: u64,
248 pub idempotency: IdempotencyOutcome,
249 pub audit: AuditEvent,
250}
251
252#[derive(Debug, Clone, Serialize, Deserialize)]
254pub struct BackupManifest {
255 pub schema: SchemaVersion,
256 pub backup_id: String,
257 pub env_id: EnvId,
258 pub created_at: DateTime<Utc>,
259 pub generation: u64,
260 pub integrity: StateIntegrity,
261 pub size_bytes: u64,
262}
263
264#[derive(Debug, Clone, Serialize, Deserialize)]
277pub struct BackupArtifact {
278 pub schema: SchemaVersion,
279 pub manifest: BackupManifest,
280 pub snapshot: Value,
282 pub snapshot_digest: String,
284}
285
286#[derive(Debug, Clone, Serialize, Deserialize)]
295pub struct ImportRequest {
296 pub artifact: BackupArtifact,
297}
298
299#[derive(Debug, Clone, Serialize, Deserialize)]
303pub struct ImportOutcome {
304 pub imported_generation: u64,
305 pub integrity: StateIntegrity,
306}
307
308impl ImportOutcome {
309 pub fn etag(&self) -> StateEtag {
311 StateEtag::from_integrity(&self.integrity)
312 }
313}
314
315#[derive(Debug, Clone, Serialize, Deserialize)]
323pub struct RestoreRequest {
324 pub backup_id: String,
325 pub precondition: Precondition,
326}
327
328impl RestoreRequest {
329 pub fn validate(&self) -> Result<(), RemoteContractError> {
331 if !self.precondition.is_conditional() {
332 return Err(RemoteContractError::UnconditionalRestore);
333 }
334 Ok(())
335 }
336}
337
338#[derive(Debug, Clone, Serialize, Deserialize)]
342pub struct RestoreOutcome {
343 pub restored_generation: u64,
344 pub integrity: StateIntegrity,
345}
346
347impl RestoreOutcome {
348 pub fn etag(&self) -> StateEtag {
350 StateEtag::from_integrity(&self.integrity)
351 }
352}
353
354#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
360#[serde(tag = "status", rename_all = "snake_case")]
361pub enum ReconcileCompletion {
362 Succeeded { applied: u32, pruned: u32 },
364 Failed { error: String },
366}
367
368#[derive(Debug, Clone, Serialize, Deserialize)]
382#[serde(deny_unknown_fields)]
383pub struct ReconcileCompletionRequest {
384 pub authorized_generation: u64,
386 pub authorized_etag: StateEtag,
389 pub completion: ReconcileCompletion,
391}
392
393#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Error)]
396#[serde(tag = "kind", rename_all = "kebab-case")]
397pub enum RemoteStoreError {
398 #[error("precondition failed (stale generation/etag)")]
400 PreconditionFailed(ConcurrencyConflict),
401 #[error("precondition required: {detail}")]
403 PreconditionRequired { detail: String },
404 #[error("idempotency conflict: {reason}")]
406 IdempotencyConflict { reason: String },
407 #[error("unauthorized: {reason} (policy `{policy}`)")]
409 Unauthorized { policy: String, reason: String },
410 #[error("environment not found")]
412 NotFound,
413 #[error("already exists: {detail}")]
418 AlreadyExists { detail: String },
419 #[error("invalid request: {detail}")]
423 InvalidRequest { detail: String },
424 #[error("conflict: {detail}")]
430 Conflict { detail: String },
431 #[error("dependent not found: {detail}")]
436 DependentNotFound { detail: String },
437 #[error("health gate failed for revision `{revision_id}`: {message}")]
442 HealthGateFailed {
443 revision_id: crate::ids::RevisionId,
444 failed_checks: Vec<crate::engine::HealthCheckId>,
445 message: String,
446 },
447 #[error("integrity mismatch: expected {expected}, computed {actual}")]
449 IntegrityMismatch { expected: String, actual: String },
450 #[error("not yet implemented: {detail}")]
452 NotYetImplemented { detail: String },
453 #[error("internal store error: {message}")]
455 Internal { message: String },
456}
457
458impl RemoteStoreError {
459 pub fn http_status(&self) -> u16 {
461 match self {
462 Self::PreconditionFailed(_) => 412,
463 Self::PreconditionRequired { .. } => 428,
464 Self::IdempotencyConflict { .. } => 409,
465 Self::Unauthorized { .. } => 403,
466 Self::NotFound => 404,
467 Self::AlreadyExists { .. } => 409,
468 Self::InvalidRequest { .. } => 400,
469 Self::Conflict { .. } => 409,
470 Self::DependentNotFound { .. } => 404,
471 Self::HealthGateFailed { .. } => 422,
472 Self::IntegrityMismatch { .. } => 422,
473 Self::NotYetImplemented { .. } => 501,
474 Self::Internal { .. } => 500,
475 }
476 }
477}
478
479impl From<PreconditionError> for RemoteStoreError {
480 fn from(err: PreconditionError) -> Self {
481 match err {
482 PreconditionError::Required => RemoteStoreError::PreconditionRequired {
483 detail: PreconditionError::Required.to_string(),
484 },
485 PreconditionError::Conflict(conflict) => RemoteStoreError::PreconditionFailed(conflict),
486 }
487 }
488}
489
490impl From<AuditDecision> for Result<(), RemoteStoreError> {
491 fn from(decision: AuditDecision) -> Self {
493 match decision {
494 AuditDecision::Allow { .. } => Ok(()),
495 AuditDecision::Deny { policy, reason } => {
496 Err(RemoteStoreError::Unauthorized { policy, reason })
497 }
498 }
499 }
500}
501
502#[derive(Debug, Clone, PartialEq, Eq, Error)]
503pub enum RemoteContractError {
504 #[error("idempotency key must not be empty")]
505 EmptyIdempotencyKey,
506 #[error("restore requires a precondition that pins prior state")]
507 UnconditionalRestore,
508}
509
510#[cfg(test)]
511mod tests {
512 use super::*;
513 use crate::audit::AuditResult;
514
515 fn etag(s: &str) -> StateEtag {
516 StateEtag(s.to_string())
517 }
518
519 fn sample_response(etag_value: &str, generation: u64) -> MutationResponse {
520 MutationResponse {
521 etag: etag(etag_value),
522 generation,
523 idempotency: IdempotencyOutcome::Applied,
524 audit: AuditEvent {
525 schema: SchemaVersion::AUDIT_EVENT_V1.into(),
526 event_id: "01JTKW5B4W4Q5Y1CQW93F7S5VH".to_string(),
527 ts: "2026-05-20T00:00:00Z".parse().unwrap(),
528 actor: Actor {
529 kind: "local-user".to_string(),
530 user: Some("tester".to_string()),
531 uid: Some(1000),
532 },
533 env_id: "local".to_string(),
534 noun: "traffic".to_string(),
535 verb: "set".to_string(),
536 target: serde_json::json!({"env": "local"}),
537 previous_generation: Some(generation.saturating_sub(1)),
538 new_generation: Some(generation),
539 idempotency_key: Some("k1".to_string()),
540 authorization: AuditDecision::Allow {
541 policy: "local-only".to_string(),
542 reason: "ok".to_string(),
543 },
544 result: AuditResult::Ok,
545 },
546 }
547 }
548
549 #[test]
550 fn etag_derives_from_content_hash() {
551 let resource = serde_json::json!({"generation": 1, "name": "local"});
552 let tag = StateEtag::of(&resource).unwrap();
553 assert_eq!(tag.0, StateIntegrity::sha256_of(&resource).unwrap().digest);
554 assert_eq!(tag.header_value(), format!("\"{}\"", tag.0));
555 }
556
557 #[test]
558 fn precondition_empty_is_rejected_not_blindly_passed() {
559 assert!(!Precondition::default().is_conditional());
560 let err = Precondition::default().check(&etag("abc"), 7).unwrap_err();
561 assert_eq!(err, PreconditionError::Required);
562 let mapped: RemoteStoreError = err.into();
563 assert_eq!(mapped.http_status(), 428);
564 }
565
566 #[test]
567 fn precondition_matching_etag_and_generation_passes() {
568 let pre = Precondition::matching(etag("abc"), 7);
569 assert!(pre.is_conditional());
570 assert!(pre.check(&etag("abc"), 7).is_ok());
571 }
572
573 #[test]
574 fn precondition_generation_only_is_conditional() {
575 let pre = Precondition {
576 if_match: None,
577 expected_generation: Some(7),
578 };
579 assert!(pre.is_conditional());
580 assert!(pre.check(&etag("anything"), 7).is_ok());
581 }
582
583 #[test]
584 fn precondition_stale_generation_conflicts() {
585 let pre = Precondition::matching(etag("abc"), 6);
586 let PreconditionError::Conflict(conflict) = pre.check(&etag("abc"), 7).unwrap_err() else {
587 panic!("expected a conflict");
588 };
589 assert_eq!(conflict.expected_generation, Some(6));
590 assert_eq!(conflict.actual_generation, 7);
591 }
592
593 #[test]
594 fn precondition_stale_etag_conflicts() {
595 let pre = Precondition::matching(etag("old"), 7);
596 let PreconditionError::Conflict(conflict) = pre.check(&etag("new"), 7).unwrap_err() else {
597 panic!("expected a conflict");
598 };
599 assert_eq!(conflict.expected_etag.as_deref(), Some("old"));
600 assert_eq!(conflict.actual_etag, "new");
601 }
602
603 #[test]
604 fn restore_request_requires_conditional_precondition() {
605 let blind = RestoreRequest {
606 backup_id: "b1".to_string(),
607 precondition: Precondition::default(),
608 };
609 assert_eq!(
610 blind.validate().unwrap_err(),
611 RemoteContractError::UnconditionalRestore
612 );
613
614 let guarded = RestoreRequest {
615 backup_id: "b1".to_string(),
616 precondition: Precondition::matching(etag("abc"), 3),
617 };
618 assert!(guarded.validate().is_ok());
619 }
620
621 #[test]
622 fn restore_request_precondition_is_not_serde_defaulted() {
623 let err = serde_json::from_str::<RestoreRequest>(r#"{"backup_id":"b1"}"#);
626 assert!(
627 err.is_err(),
628 "missing precondition must fail to deserialize"
629 );
630 }
631
632 #[test]
633 fn idempotency_key_rejects_empty() {
634 assert!(IdempotencyKey::new(" ").is_err());
635 assert_eq!(IdempotencyKey::new("k1").unwrap().as_str(), "k1");
636 }
637
638 #[test]
639 fn idempotency_key_deserializes_through_validation() {
640 assert!(serde_json::from_str::<IdempotencyKey>("\"\"").is_err());
641 let key: IdempotencyKey = serde_json::from_str("\"01JABC\"").unwrap();
642 assert_eq!(key.as_str(), "01JABC");
643 }
644
645 #[test]
646 fn idempotency_same_body_replays_different_body_conflicts() {
647 let body = serde_json::json!({"split": [{"rev": "a", "bps": 10000}]});
648 let record = IdempotencyRecord {
649 key: IdempotencyKey::new("k1").unwrap(),
650 request_fingerprint: IdempotencyRecord::fingerprint(&body).unwrap(),
651 response: sample_response("abc", 3),
652 stored_at: Utc::now(),
653 };
654
655 let same = IdempotencyRecord::fingerprint(&body).unwrap();
656 assert!(matches!(
657 record.match_request(&same),
658 IdempotencyReplay::Replay(_)
659 ));
660
661 let other = serde_json::json!({"split": [{"rev": "b", "bps": 10000}]});
662 let other_fp = IdempotencyRecord::fingerprint(&other).unwrap();
663 assert!(matches!(
664 record.match_request(&other_fp),
665 IdempotencyReplay::Conflict { .. }
666 ));
667 }
668
669 #[test]
670 fn idempotency_replay_returns_original_response_and_audit_verbatim() {
671 let body = serde_json::json!({"split": [{"rev": "a", "bps": 10000}]});
672 let original = sample_response("abc", 3);
673 let record = IdempotencyRecord {
674 key: IdempotencyKey::new("k1").unwrap(),
675 request_fingerprint: IdempotencyRecord::fingerprint(&body).unwrap(),
676 response: original.clone(),
677 stored_at: Utc::now(),
678 };
679
680 let same = IdempotencyRecord::fingerprint(&body).unwrap();
681 let IdempotencyReplay::Replay(replayed) = record.match_request(&same) else {
682 panic!("expected a replay");
683 };
684 assert_eq!(replayed.etag, original.etag);
685 assert_eq!(replayed.generation, original.generation);
686 assert_eq!(replayed.audit.event_id, original.audit.event_id);
687 assert_eq!(replayed.audit.verb, "set");
688 let json = serde_json::to_string(&record).unwrap();
691 let back: IdempotencyRecord = serde_json::from_str(&json).unwrap();
692 assert_eq!(back.response.audit.event_id, original.audit.event_id);
693 }
694
695 #[test]
696 fn deny_decision_maps_to_unauthorized() {
697 let denied = AuditDecision::Deny {
698 policy: "local-only".to_string(),
699 reason: "non-local".to_string(),
700 };
701 let result: Result<(), RemoteStoreError> = denied.into();
702 let err = result.unwrap_err();
703 assert_eq!(err.http_status(), 403);
704 assert!(matches!(err, RemoteStoreError::Unauthorized { .. }));
705
706 let allowed = AuditDecision::Allow {
707 policy: "local-only".to_string(),
708 reason: "ok".to_string(),
709 };
710 let result: Result<(), RemoteStoreError> = allowed.into();
711 assert!(result.is_ok());
712 }
713
714 #[test]
715 fn error_status_mapping_is_stable() {
716 assert_eq!(
717 RemoteStoreError::PreconditionFailed(ConcurrencyConflict {
718 expected_etag: None,
719 actual_etag: "x".to_string(),
720 expected_generation: None,
721 actual_generation: 1,
722 })
723 .http_status(),
724 412
725 );
726 assert_eq!(
727 RemoteStoreError::PreconditionRequired {
728 detail: "x".to_string()
729 }
730 .http_status(),
731 428
732 );
733 assert_eq!(
734 RemoteStoreError::IdempotencyConflict {
735 reason: "x".to_string()
736 }
737 .http_status(),
738 409
739 );
740 assert_eq!(RemoteStoreError::NotFound.http_status(), 404);
741 assert_eq!(
742 RemoteStoreError::AlreadyExists {
743 detail: "x".to_string()
744 }
745 .http_status(),
746 409
747 );
748 assert_eq!(
749 RemoteStoreError::InvalidRequest {
750 detail: "x".to_string()
751 }
752 .http_status(),
753 400
754 );
755 assert_eq!(
756 RemoteStoreError::IntegrityMismatch {
757 expected: "a".to_string(),
758 actual: "b".to_string()
759 }
760 .http_status(),
761 422
762 );
763 assert_eq!(
764 RemoteStoreError::NotYetImplemented {
765 detail: "x".to_string()
766 }
767 .http_status(),
768 501
769 );
770 assert_eq!(
771 RemoteStoreError::Internal {
772 message: "x".to_string()
773 }
774 .http_status(),
775 500
776 );
777 assert_eq!(
778 RemoteStoreError::Conflict {
779 detail: "x".to_string()
780 }
781 .http_status(),
782 409
783 );
784 assert_eq!(
785 RemoteStoreError::DependentNotFound {
786 detail: "x".to_string()
787 }
788 .http_status(),
789 404
790 );
791 assert_eq!(
792 RemoteStoreError::HealthGateFailed {
793 revision_id: crate::ids::RevisionId::new(),
794 failed_checks: Vec::new(),
795 message: "x".to_string()
796 }
797 .http_status(),
798 422
799 );
800 }
801
802 #[test]
803 fn remote_store_error_round_trips_tagged() {
804 let err = RemoteStoreError::Unauthorized {
805 policy: "local-only".to_string(),
806 reason: "nope".to_string(),
807 };
808 let json = serde_json::to_value(&err).unwrap();
809 assert_eq!(json["kind"], "unauthorized");
810 let back: RemoteStoreError = serde_json::from_value(json).unwrap();
811 assert_eq!(back, err);
812 }
813
814 #[test]
815 fn health_gate_failed_round_trips_tagged() {
816 let err = RemoteStoreError::HealthGateFailed {
817 revision_id: crate::ids::RevisionId::new(),
818 failed_checks: vec![crate::engine::HealthCheckId::RouteTable],
819 message: "route table invalid".to_string(),
820 };
821 let json = serde_json::to_value(&err).unwrap();
822 assert_eq!(json["kind"], "health-gate-failed");
823 assert_eq!(json["failed_checks"][0], "route-table");
824 let back: RemoteStoreError = serde_json::from_value(json).unwrap();
825 assert_eq!(back, err);
826 }
827}