lenso_service/
delivery_console_projection.rs

1use std::collections::{BTreeMap, BTreeSet};
2
3use schemars::JsonSchema;
4use serde::{Deserialize, Serialize, de::DeserializeOwned};
5use serde_json::Value;
6use utoipa::ToSchema;
7
8use crate::extraction_input_digest;
9use crate::{
10    CANARY_DECISION_PROTOCOL, CANARY_PLAN_PROTOCOL, CONFIG_ACTIVATION_RECEIPT_PROTOCOL,
11    CONFIG_REVISION_PROTOCOL, CanaryDecision, CanaryPlan, ConfigActivationReceipt, ConfigRevision,
12    DEPLOYMENT_OBSERVATION_PROTOCOL, DEPLOYMENT_PLAN_PROTOCOL, DEPLOYMENT_RECEIPT_PROTOCOL,
13    DeploymentObservation, DeploymentPlan, DeploymentReceipt, EDGE_CONTRACT_PROTOCOL,
14    ENVIRONMENT_VERIFICATION_PROTOCOL, EdgeContract, EnvironmentVerification,
15    GATEWAY_OBSERVATION_PROTOCOL, GATEWAY_PLAN_PROTOCOL, GatewayConfigurationPlan,
16    GatewayObservation, POLICY_EVIDENCE_PROTOCOL, PROMOTION_APPROVAL_PROTOCOL,
17    PROMOTION_PLAN_PROTOCOL, PROMOTION_RECEIPT_PROTOCOL, PolicyEvidence, PromotionApproval,
18    PromotionPlan, PromotionReceipt, RELEASE_TRUST_EVIDENCE_PROTOCOL,
19    RELIABILITY_OBSERVATION_PROTOCOL, ROLLBACK_PLAN_PROTOCOL, ROLLBACK_RECEIPT_PROTOCOL,
20    ReleaseTrustEvidence, ReliabilityObservation, RollbackPlan, RollbackReceipt,
21    SERVICE_RELEASE_PROTOCOL, ServiceRelease, canary_decision_integrity_is_valid,
22    canary_plan_integrity_is_valid, config_revision_integrity_is_valid,
23    deployment_observation_content_integrity_is_valid, deployment_plan_integrity_is_valid,
24    edge_contract_integrity_is_valid, environment_verification_integrity_is_valid,
25    gateway_observation_content_integrity_is_valid, gateway_plan_integrity_is_valid,
26    policy_evidence_integrity_is_valid, promotion_plan_integrity_is_valid,
27    rollback_plan_integrity_is_valid, secret_reference_metadata_is_safe,
28    service_release_integrity_is_valid,
29};
30use crate::{
31    CONTRACT_RETIREMENT_PLAN_PROTOCOL, CONTRACT_RETIREMENT_RECEIPT_PROTOCOL,
32    ContractRetirementPlan, ContractRetirementReceipt, DELIVERY_FAILURE_RECOVERY_PROTOCOL,
33    DISASTER_RECOVERY_EVIDENCE_PROTOCOL, DeliveryFailureRecoveryEvidence, DisasterRecoveryEvidence,
34    GA_SUPPORT_MANIFEST_PROTOCOL, GaSupportManifest, PERFORMANCE_PROFILE_PROTOCOL,
35    PerformanceProfile, SECURITY_REVIEW_PROTOCOL, SERVICE_RESTORE_EVIDENCE_PROTOCOL,
36    SUPPORT_ENVELOPE_PROTOCOL, SecurityReviewEvidence, ServiceRestoreEvidence, SupportEnvelope,
37    contract_retirement_plan_integrity_is_valid, contract_retirement_receipt_integrity_is_valid,
38    delivery_failure_recovery_integrity_is_valid, disaster_recovery_evidence_integrity_is_valid,
39    ga_support_manifest_integrity_valid, performance_profile_integrity_is_valid,
40    security_review_integrity_is_valid, service_restore_integrity_is_valid,
41    support_envelope_integrity_is_valid,
42};
43
44pub const DELIVERY_CONSOLE_PROJECTION_PROTOCOL: &str = "lenso.delivery-console.v1";
45pub const DELIVERY_ARTIFACT_BATCH_PROTOCOL: &str = "lenso.delivery-artifact-batch.v1";
46
47#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
48#[serde(rename_all = "camelCase")]
49pub struct DeliveryConsoleArtifacts {
50    #[serde(default)]
51    pub artifacts: Vec<Value>,
52}
53
54#[derive(
55    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema, ToSchema,
56)]
57#[serde(rename_all = "snake_case")]
58pub enum DeliveryConsoleState {
59    Planned,
60    Blocked,
61    Staged,
62    Approved,
63    Canary,
64    Converging,
65    Ready,
66    RollingBack,
67    RolledBack,
68    Paused,
69    InterventionRequired,
70}
71
72#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
73#[serde(rename_all = "camelCase")]
74pub struct DeliveryConsoleRelease {
75    pub service_id: String,
76    pub release_id: String,
77    pub release_digest: String,
78}
79
80#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
81#[serde(rename_all = "camelCase")]
82pub struct DeliveryConsoleSupplyChainWorkload {
83    pub workload_id: String,
84    pub artifact_digest: String,
85    pub signature_status: String,
86    pub sbom_reference: String,
87    pub provenance_reference: String,
88    pub provenance_subject_matches: bool,
89}
90
91#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
92#[serde(rename_all = "camelCase")]
93pub struct DeliveryConsolePolicy {
94    pub evidence_id: String,
95    pub pack_id: String,
96    pub decision: String,
97}
98
99#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
100#[serde(rename_all = "camelCase")]
101pub struct DeliveryConsoleSecretReference {
102    pub reference_id: String,
103    pub provider: String,
104    pub purpose: String,
105    pub scope: String,
106    pub status: String,
107    #[serde(default, skip_serializing_if = "Option::is_none")]
108    pub rotation_revision: Option<String>,
109}
110
111#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
112#[serde(rename_all = "camelCase")]
113pub struct DeliveryConsoleConfiguration {
114    #[serde(default, skip_serializing_if = "Option::is_none")]
115    pub desired_revision_id: Option<String>,
116    #[serde(default, skip_serializing_if = "Option::is_none")]
117    pub active_revision_id: Option<String>,
118    #[serde(default, skip_serializing_if = "Option::is_none")]
119    pub previous_revision_id: Option<String>,
120    pub drifted: bool,
121    #[serde(default)]
122    pub secret_references: Vec<DeliveryConsoleSecretReference>,
123}
124
125#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
126#[serde(rename_all = "camelCase")]
127pub struct DeliveryConsoleDeployment {
128    pub environment: String,
129    pub desired_release_id: String,
130    pub observed_release_id: String,
131    pub config_revision_id: String,
132    pub drifted: bool,
133    pub fresh: bool,
134}
135
136#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
137#[serde(rename_all = "camelCase")]
138pub struct DeliveryConsoleAdapterDrift {
139    pub environment: String,
140    pub drifted: bool,
141    pub fresh: bool,
142    #[serde(default)]
143    pub next_actions: Vec<String>,
144}
145
146#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
147#[serde(rename_all = "camelCase")]
148pub struct DeliveryConsoleEdge {
149    pub contract_id: String,
150    #[serde(default)]
151    pub public_routes: Vec<String>,
152}
153
154#[derive(
155    Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema, ToSchema,
156)]
157#[serde(rename_all = "camelCase")]
158pub struct DeliveryConsoleIssue {
159    pub code: String,
160    pub message: String,
161    #[serde(default)]
162    pub evidence_references: Vec<String>,
163    pub remediation: String,
164    #[serde(default)]
165    pub next_actions: Vec<String>,
166}
167
168#[derive(
169    Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema, ToSchema,
170)]
171#[serde(rename_all = "camelCase")]
172pub struct DeliveryConsoleTimelineEntry {
173    pub protocol: String,
174    pub artifact_id: String,
175    pub state: String,
176    #[serde(default)]
177    pub evidence_references: Vec<String>,
178}
179
180#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
181#[serde(rename_all = "camelCase")]
182pub struct DeliveryConsoleDependencyObservation {
183    pub dependency_id: String,
184    pub available: bool,
185    #[serde(default, skip_serializing_if = "Option::is_none")]
186    pub active_degraded_mode: Option<String>,
187}
188
189#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
190#[serde(rename_all = "camelCase")]
191pub struct DeliveryConsoleCanaryObservation {
192    pub observation_id: String,
193    pub observed_revision: u64,
194    pub fresh: bool,
195    pub observation_window_seconds: u64,
196    pub sample_count: u64,
197    pub generic_process_healthy: bool,
198    #[serde(default)]
199    pub workload_readiness: BTreeMap<String, bool>,
200    #[serde(default)]
201    pub workload_liveness: BTreeMap<String, bool>,
202    pub availability_basis_points: Option<u32>,
203    pub latency_p99_ms: Option<u64>,
204    pub error_budget_used_basis_points: Option<u32>,
205    pub queue_backlog: Option<u64>,
206    pub workflow_backlog: Option<u64>,
207    pub timer_lag_ms: Option<u64>,
208    pub retry_exhaustion: Option<u64>,
209    pub compensation_pressure: Option<u64>,
210    #[serde(default)]
211    pub dependencies: Vec<DeliveryConsoleDependencyObservation>,
212    #[serde(default)]
213    pub failure_domains: BTreeMap<String, bool>,
214    pub scaling_check_passed: Option<bool>,
215    pub disruption_check_passed: Option<bool>,
216    pub availability_check_passed: Option<bool>,
217    #[serde(default)]
218    pub evidence_references: Vec<String>,
219}
220
221#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
222#[serde(rename_all = "camelCase")]
223pub struct DeliveryConsoleGaEvidence {
224    pub protocol: String,
225    pub evidence_id: String,
226    pub status: String,
227    pub stale: bool,
228    #[serde(default)]
229    pub subjects: BTreeMap<String, String>,
230    #[serde(default)]
231    pub details: BTreeMap<String, Value>,
232    #[serde(default)]
233    pub issue_codes: Vec<String>,
234    #[serde(default)]
235    pub next_actions: Vec<String>,
236}
237
238#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
239#[serde(rename_all = "camelCase")]
240pub struct DeliveryConsoleGaOperations {
241    #[serde(default, skip_serializing_if = "Option::is_none")]
242    pub support_manifest: Option<DeliveryConsoleGaEvidence>,
243    #[serde(default)]
244    pub delivery_recovery: Vec<DeliveryConsoleGaEvidence>,
245    #[serde(default, skip_serializing_if = "Option::is_none")]
246    pub restore: Option<DeliveryConsoleGaEvidence>,
247    #[serde(default, skip_serializing_if = "Option::is_none")]
248    pub disaster_recovery: Option<DeliveryConsoleGaEvidence>,
249    #[serde(default, skip_serializing_if = "Option::is_none")]
250    pub performance: Option<DeliveryConsoleGaEvidence>,
251    #[serde(default, skip_serializing_if = "Option::is_none")]
252    pub support_envelope: Option<DeliveryConsoleGaEvidence>,
253    #[serde(default, skip_serializing_if = "Option::is_none")]
254    pub security_review: Option<DeliveryConsoleGaEvidence>,
255    #[serde(default)]
256    pub contract_lifecycle: Vec<DeliveryConsoleGaEvidence>,
257}
258
259#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, ToSchema)]
260#[serde(rename_all = "camelCase")]
261pub struct DeliveryConsoleProjection {
262    pub protocol: String,
263    pub projection_digest: String,
264    pub state: DeliveryConsoleState,
265    #[serde(default, skip_serializing_if = "Option::is_none")]
266    pub release: Option<DeliveryConsoleRelease>,
267    #[serde(default)]
268    pub supply_chain: Vec<DeliveryConsoleSupplyChainWorkload>,
269    #[serde(default, skip_serializing_if = "Option::is_none")]
270    pub policy: Option<DeliveryConsolePolicy>,
271    pub configuration: DeliveryConsoleConfiguration,
272    #[serde(default)]
273    pub deployments: Vec<DeliveryConsoleDeployment>,
274    #[serde(default, skip_serializing_if = "Option::is_none")]
275    pub edge: Option<DeliveryConsoleEdge>,
276    #[serde(default)]
277    pub adapter_drift: Vec<DeliveryConsoleAdapterDrift>,
278    #[serde(default)]
279    pub promotion_history: Vec<DeliveryConsoleTimelineEntry>,
280    #[serde(default)]
281    pub canary_timeline: Vec<DeliveryConsoleTimelineEntry>,
282    #[serde(default)]
283    pub canary_observations: Vec<DeliveryConsoleCanaryObservation>,
284    #[serde(default)]
285    pub rollback_timeline: Vec<DeliveryConsoleTimelineEntry>,
286    #[serde(default)]
287    pub issues: Vec<DeliveryConsoleIssue>,
288    #[serde(default)]
289    pub next_actions: Vec<String>,
290    #[serde(default)]
291    pub runtime_story_references: Vec<String>,
292    #[serde(default)]
293    pub ga_operations: DeliveryConsoleGaOperations,
294    pub read_only: bool,
295    #[serde(default)]
296    pub apply_actions: Vec<String>,
297}
298
299#[must_use]
300pub fn project_delivery_console(input: DeliveryConsoleArtifacts) -> DeliveryConsoleProjection {
301    // Artifact order is the append-only ledger order. Keep it intact so the
302    // newest state is selected by recording sequence rather than hash-like IDs.
303    let release_artifact = latest(&input.artifacts, "lenso.service-release.v1");
304    let trust_artifact = latest(&input.artifacts, "lenso.release-trust-evidence.v1");
305    let policy_artifact = input
306        .artifacts
307        .iter()
308        .rev()
309        .find(|artifact| protocol(artifact).contains("policy-evidence"));
310    let config_artifact = latest(&input.artifacts, "lenso.config-revision.v1");
311    let config_receipt = latest(&input.artifacts, "lenso.config-activation-receipt.v1");
312    let release = release_artifact.map(|artifact| DeliveryConsoleRelease {
313        service_id: text(artifact, "serviceId").unwrap_or_else(|| "unknown".to_owned()),
314        release_id: text(artifact, "releaseId").unwrap_or_else(|| "unknown".to_owned()),
315        release_digest: text(artifact, "releaseDigest").unwrap_or_else(|| "unknown".to_owned()),
316    });
317    let signature_status = trust_artifact
318        .and_then(|artifact| array(artifact, "signatures").into_iter().next())
319        .and_then(|signature| text(signature, "status"))
320        .unwrap_or_else(|| "unknown".to_owned());
321    let trust_workloads = trust_artifact
322        .map(|artifact| array(artifact, "workloads"))
323        .unwrap_or_default();
324    let mut supply_chain = release_artifact
325        .map(|artifact| array(artifact, "workloads"))
326        .unwrap_or_default()
327        .into_iter()
328        .map(|workload| {
329            let workload_id = text(workload, "workloadId").unwrap_or_else(|| "unknown".to_owned());
330            let trust = trust_workloads
331                .iter()
332                .find(|item| text(item, "workloadId").as_deref() == Some(workload_id.as_str()));
333            DeliveryConsoleSupplyChainWorkload {
334                workload_id,
335                artifact_digest: text(workload, "artifactDigest")
336                    .unwrap_or_else(|| "unknown".to_owned()),
337                signature_status: signature_status.clone(),
338                sbom_reference: nested_text(workload, "sbom", "reference")
339                    .unwrap_or_else(|| "missing".to_owned()),
340                provenance_reference: nested_text(workload, "provenance", "reference")
341                    .unwrap_or_else(|| "missing".to_owned()),
342                provenance_subject_matches: trust
343                    .and_then(|item| item.get("provenanceSubjectMatches"))
344                    .and_then(Value::as_bool)
345                    .unwrap_or(false),
346            }
347        })
348        .collect::<Vec<_>>();
349    supply_chain.sort_by(|left, right| left.workload_id.cmp(&right.workload_id));
350    let policy = policy_artifact.map(|artifact| DeliveryConsolePolicy {
351        evidence_id: text(artifact, "evidenceId").unwrap_or_else(|| "unknown".to_owned()),
352        pack_id: text(artifact, "packId").unwrap_or_else(|| "unknown".to_owned()),
353        decision: text(artifact, "decision").unwrap_or_else(|| "unknown".to_owned()),
354    });
355    let desired_revision_id = config_artifact.and_then(|artifact| text(artifact, "revisionId"));
356    let active_revision_id =
357        config_receipt.and_then(|artifact| match text(artifact, "activation").as_deref() {
358            Some("active" | "rolled_back") => text(artifact, "targetRevisionId"),
359            Some("staged") => text(artifact, "previousRevisionId"),
360            _ => None,
361        });
362    let previous_revision_id =
363        config_receipt.and_then(|artifact| text(artifact, "previousRevisionId"));
364    let mut secret_references = config_artifact
365        .map(|artifact| array(artifact, "secretReferences"))
366        .unwrap_or_default()
367        .into_iter()
368        .map(|reference| DeliveryConsoleSecretReference {
369            reference_id: text(reference, "referenceId").unwrap_or_else(|| "unknown".to_owned()),
370            provider: text(reference, "provider").unwrap_or_else(|| "unknown".to_owned()),
371            purpose: text(reference, "purpose").unwrap_or_else(|| "unknown".to_owned()),
372            scope: text(reference, "scope").unwrap_or_else(|| "unknown".to_owned()),
373            status: text(reference, "status").unwrap_or_else(|| "unknown".to_owned()),
374            rotation_revision: reference
375                .get("metadata")
376                .and_then(|metadata| text(metadata, "rotationRevision")),
377        })
378        .collect::<Vec<_>>();
379    secret_references.sort_by(|left, right| left.reference_id.cmp(&right.reference_id));
380    let mut latest_deployments = std::collections::BTreeMap::new();
381    for artifact in input
382        .artifacts
383        .iter()
384        .filter(|artifact| protocol(artifact) == "lenso.deployment-observation.v1")
385    {
386        let deployment = DeliveryConsoleDeployment {
387            environment: text(artifact, "environment").unwrap_or_else(|| "unknown".to_owned()),
388            desired_release_id: text(artifact, "desiredReleaseId")
389                .unwrap_or_else(|| "unknown".to_owned()),
390            observed_release_id: text(artifact, "observedReleaseId")
391                .unwrap_or_else(|| "unknown".to_owned()),
392            config_revision_id: text(artifact, "configRevisionId")
393                .unwrap_or_else(|| "unknown".to_owned()),
394            drifted: boolean(artifact, "drifted"),
395            fresh: boolean(artifact, "fresh"),
396        };
397        latest_deployments.insert(deployment.environment.clone(), deployment);
398    }
399    let mut deployments = latest_deployments.into_values().collect::<Vec<_>>();
400    deployments.sort_by(|left, right| left.environment.cmp(&right.environment));
401    let adapter_drift = deployments
402        .iter()
403        .map(|deployment| DeliveryConsoleAdapterDrift {
404            environment: deployment.environment.clone(),
405            drifted: deployment.drifted,
406            fresh: deployment.fresh,
407            next_actions: if deployment.drifted || !deployment.fresh {
408                vec![
409                    "Refresh adapter observations and reconcile the exact Deployment plan."
410                        .to_owned(),
411                ]
412            } else {
413                Vec::new()
414            },
415        })
416        .collect::<Vec<_>>();
417    let edge = latest(&input.artifacts, "lenso.edge-contract.v1").map(|artifact| {
418        let mut public_routes = array(artifact, "routes")
419            .into_iter()
420            .filter_map(|route| text(route, "publicPath"))
421            .collect::<Vec<_>>();
422        public_routes.sort();
423        DeliveryConsoleEdge {
424            contract_id: text(artifact, "edgeContractId").unwrap_or_else(|| "unknown".to_owned()),
425            public_routes,
426        }
427    });
428    let mut timeline = input
429        .artifacts
430        .iter()
431        .filter(|artifact| is_timeline_protocol(protocol(artifact)))
432        .map(timeline_entry)
433        .collect::<Vec<_>>();
434    let mut seen_timeline = BTreeSet::new();
435    timeline.retain(|entry| seen_timeline.insert(entry.clone()));
436    let promotion_history = filter_timeline(&timeline, "promotion");
437    let canary_timeline = filter_timeline(&timeline, "canary");
438    let canary_observations = input
439        .artifacts
440        .iter()
441        .filter(|artifact| protocol(artifact) == "lenso.reliability-observation.v1")
442        .filter_map(canary_observation)
443        .collect::<Vec<_>>();
444    let rollback_timeline = filter_timeline(&timeline, "rollback");
445    let mut issues = input
446        .artifacts
447        .iter()
448        .flat_map(issues_from)
449        .collect::<Vec<_>>();
450    issues.sort();
451    issues.dedup();
452    let mut next_actions = issues
453        .iter()
454        .flat_map(|issue| issue.next_actions.iter().cloned())
455        .collect::<Vec<_>>();
456    next_actions.sort();
457    next_actions.dedup();
458    let mut runtime_story_references = BTreeSet::new();
459    for artifact in &input.artifacts {
460        collect_runtime_story_references(artifact, &mut runtime_story_references);
461    }
462    let configuration = DeliveryConsoleConfiguration {
463        drifted: desired_revision_id != active_revision_id
464            || deployments.iter().any(|deployment| {
465                desired_revision_id
466                    .as_deref()
467                    .is_some_and(|desired| deployment.config_revision_id != desired)
468            }),
469        desired_revision_id,
470        active_revision_id,
471        previous_revision_id,
472        secret_references,
473    };
474    let state = derive_state(&input.artifacts, !issues.is_empty(), &deployments);
475    let ga_operations = ga_operations(&input.artifacts);
476    let mut projection = DeliveryConsoleProjection {
477        protocol: DELIVERY_CONSOLE_PROJECTION_PROTOCOL.to_owned(),
478        projection_digest: String::new(),
479        state,
480        release,
481        supply_chain,
482        policy,
483        configuration,
484        deployments,
485        edge,
486        adapter_drift,
487        promotion_history,
488        canary_timeline,
489        canary_observations,
490        rollback_timeline,
491        issues,
492        next_actions,
493        runtime_story_references: runtime_story_references.into_iter().collect(),
494        ga_operations,
495        read_only: true,
496        apply_actions: Vec::new(),
497    };
498    projection.projection_digest = digest(&projection);
499    projection
500}
501
502fn ga_operations(artifacts: &[Value]) -> DeliveryConsoleGaOperations {
503    let summaries = |protocol_name: &str| {
504        artifacts
505            .iter()
506            .filter(|artifact| protocol(artifact) == protocol_name)
507            .filter_map(ga_evidence)
508            .collect::<Vec<_>>()
509    };
510    let newest = |protocol_name: &str| summaries(protocol_name).pop();
511    let mut contract_lifecycle = summaries("lenso.contract-retirement-plan.v1");
512    contract_lifecycle.extend(summaries("lenso.contract-retirement-receipt.v2"));
513    DeliveryConsoleGaOperations {
514        support_manifest: newest("lenso.ga-support-manifest.v1"),
515        delivery_recovery: summaries("lenso.delivery-failure-recovery-evidence.v1"),
516        restore: newest("lenso.service-restore-evidence.v1"),
517        disaster_recovery: newest("lenso.disaster-recovery-evidence.v1"),
518        performance: newest("lenso.performance-profile.v1"),
519        support_envelope: newest("lenso.support-envelope.v1"),
520        security_review: newest("lenso.security-review-evidence.v1"),
521        contract_lifecycle,
522    }
523}
524
525fn ga_evidence(artifact: &Value) -> Option<DeliveryConsoleGaEvidence> {
526    let protocol = text(artifact, "protocol")?;
527    let evidence_id = [
528        "evidenceId",
529        "profileId",
530        "envelopeId",
531        "reviewId",
532        "manifestId",
533        "planId",
534        "receiptId",
535    ]
536    .into_iter()
537    .find_map(|key| text(artifact, key))
538    .unwrap_or_else(|| "unknown".to_owned());
539    let status = ["decision", "status", "outcome"]
540        .into_iter()
541        .find_map(|key| text(artifact, key))
542        .unwrap_or_else(|| "unknown".to_owned());
543    let issue_codes = array(artifact, "issues")
544        .into_iter()
545        .filter_map(|issue| text(issue, "code"))
546        .collect::<Vec<_>>();
547    let stale = issue_codes
548        .iter()
549        .any(|code| code.contains("stale") || code.contains("freshness"));
550    let subjects = [
551        "serviceId",
552        "workloadId",
553        "releaseId",
554        "releaseDigest",
555        "configRevisionId",
556        "configRevisionDigest",
557        "contractId",
558        "contractSetDigest",
559        "deploymentId",
560        "storyId",
561        "backupId",
562        "supportManifestDigest",
563        "primaryRegion",
564        "passiveRegion",
565        "phase",
566        "observedRpoMs",
567        "observedRtoMs",
568        "recoveryTimeMs",
569        "intentionalLossBoundMs",
570        "replayBoundCount",
571        "freshnessHorizonUnixMs",
572        "upgradeStatus",
573        "rollbackStatus",
574    ]
575    .into_iter()
576    .filter_map(|key| scalar_text(artifact, key).map(|value| (key.to_owned(), value)))
577    .chain(
578        [
579            ("activeConsumerCount", "activeConsumers"),
580            ("performanceBudgetCount", "budgets"),
581            ("findingCount", "findings"),
582            ("contractVersionCount", "contractVersionDigests"),
583            ("remainingStoryGapCount", "remainingStoryGaps"),
584        ]
585        .into_iter()
586        .filter_map(|(label, key)| {
587            artifact
588                .get(key)
589                .and_then(|value| match value {
590                    Value::Array(values) => Some(values.len()),
591                    Value::Object(values) => Some(values.len()),
592                    _ => None,
593                })
594                .map(|count| (label.to_owned(), count.to_string()))
595        }),
596    )
597    .collect();
598    let details = [
599        "components",
600        "manifestFormats",
601        "stateVersions",
602        "adapterVersions",
603        "combinations",
604        "upgradeEdges",
605        "activeConsumers",
606        "budgets",
607        "measurements",
608        "findings",
609        "contractVersionDigests",
610        "remainingStoryGaps",
611        "reconciliation",
612        "environmentObservation",
613    ]
614    .into_iter()
615    .filter_map(|key| {
616        artifact
617            .get(key)
618            .filter(|value| !value.is_null())
619            .cloned()
620            .map(|value| (key.to_owned(), value))
621    })
622    .collect();
623    Some(DeliveryConsoleGaEvidence {
624        protocol,
625        evidence_id,
626        status,
627        stale,
628        subjects,
629        details,
630        issue_codes,
631        next_actions: strings(artifact, "nextActions"),
632    })
633}
634
635fn scalar_text(value: &Value, key: &str) -> Option<String> {
636    value.get(key).and_then(|value| match value {
637        Value::String(value) => Some(value.clone()),
638        Value::Number(value) => Some(value.to_string()),
639        Value::Bool(value) => Some(value.to_string()),
640        _ => None,
641    })
642}
643
644fn canary_observation(artifact: &Value) -> Option<DeliveryConsoleCanaryObservation> {
645    Some(DeliveryConsoleCanaryObservation {
646        observation_id: text(artifact, "observationId")?,
647        observed_revision: artifact.get("observedRevision")?.as_u64()?,
648        fresh: boolean(artifact, "fresh"),
649        observation_window_seconds: artifact.get("observationWindowSeconds")?.as_u64()?,
650        sample_count: artifact.get("sampleCount")?.as_u64()?,
651        generic_process_healthy: boolean(artifact, "genericProcessHealthy"),
652        workload_readiness: bool_map(artifact, "workloadReadiness"),
653        workload_liveness: bool_map(artifact, "workloadLiveness"),
654        availability_basis_points: artifact
655            .get("availabilityBasisPoints")
656            .and_then(Value::as_u64)
657            .and_then(|value| u32::try_from(value).ok()),
658        latency_p99_ms: artifact.get("latencyP99Ms").and_then(Value::as_u64),
659        error_budget_used_basis_points: artifact
660            .get("errorBudgetUsedBasisPoints")
661            .and_then(Value::as_u64)
662            .and_then(|value| u32::try_from(value).ok()),
663        queue_backlog: artifact.get("queueBacklog").and_then(Value::as_u64),
664        workflow_backlog: artifact.get("workflowBacklog").and_then(Value::as_u64),
665        timer_lag_ms: artifact.get("timerLagMs").and_then(Value::as_u64),
666        retry_exhaustion: artifact.get("retryExhaustion").and_then(Value::as_u64),
667        compensation_pressure: artifact.get("compensationPressure").and_then(Value::as_u64),
668        dependencies: array(artifact, "dependencies")
669            .into_iter()
670            .filter_map(|dependency| {
671                Some(DeliveryConsoleDependencyObservation {
672                    dependency_id: text(dependency, "dependencyId")?,
673                    available: boolean(dependency, "available"),
674                    active_degraded_mode: text(dependency, "activeDegradedMode"),
675                })
676            })
677            .collect(),
678        failure_domains: bool_map(artifact, "failureDomains"),
679        scaling_check_passed: artifact.get("scalingCheckPassed").and_then(Value::as_bool),
680        disruption_check_passed: artifact
681            .get("disruptionCheckPassed")
682            .and_then(Value::as_bool),
683        availability_check_passed: artifact
684            .get("availabilityCheckPassed")
685            .and_then(Value::as_bool),
686        evidence_references: array(artifact, "evidenceReferences")
687            .into_iter()
688            .filter_map(Value::as_str)
689            .map(str::to_owned)
690            .collect(),
691    })
692}
693
694fn bool_map(artifact: &Value, field: &str) -> BTreeMap<String, bool> {
695    artifact
696        .get(field)
697        .and_then(Value::as_object)
698        .into_iter()
699        .flatten()
700        .filter_map(|(key, value)| value.as_bool().map(|value| (key.clone(), value)))
701        .collect()
702}
703
704/// Persist one immutable production-delivery artifact for the read-only projection.
705pub async fn record_delivery_artifact(
706    pool: &sqlx::PgPool,
707    delivery_id: &str,
708    artifact: &Value,
709) -> Result<(), sqlx::Error> {
710    let persisted = persisted_delivery_artifact(artifact)?;
711    sqlx::query(
712        r#"
713        insert into platform.delivery_artifacts
714            (delivery_id, artifact_id, protocol, artifact_digest, artifact_json)
715        values ($1, $2, $3, $4, $5)
716        on conflict (delivery_id, artifact_id, artifact_digest) do nothing
717        "#,
718    )
719    .bind(delivery_id)
720    .bind(artifact_id(&persisted))
721    .bind(protocol(&persisted))
722    .bind(digest(artifact))
723    .bind(&persisted)
724    .execute(pool)
725    .await?;
726    Ok(())
727}
728
729/// Compute the canonical subject signed by the provider that records a delivery batch.
730#[must_use]
731pub fn delivery_artifact_batch_subject(delivery_id: &str, artifacts: &[Value]) -> String {
732    extraction_input_digest(
733        serde_json::to_vec(&(DELIVERY_ARTIFACT_BATCH_PROTOCOL, delivery_id, artifacts))
734            .expect("delivery artifact batches must serialize"),
735    )
736}
737
738/// Validate and atomically persist a signed production-delivery artifact batch.
739pub async fn record_delivery_artifacts(
740    pool: &sqlx::PgPool,
741    delivery_id: &str,
742    artifacts: &[Value],
743) -> Result<(), sqlx::Error> {
744    let persisted = artifacts
745        .iter()
746        .map(persisted_delivery_artifact)
747        .collect::<Result<Vec<_>, _>>()?;
748    let mut transaction = pool.begin().await?;
749    for (artifact, persisted) in artifacts.iter().zip(&persisted) {
750        sqlx::query(
751            r#"
752            insert into platform.delivery_artifacts
753                (delivery_id, artifact_id, protocol, artifact_digest, artifact_json)
754            values ($1, $2, $3, $4, $5)
755            on conflict (delivery_id, artifact_id, artifact_digest) do nothing
756            "#,
757        )
758        .bind(delivery_id)
759        .bind(artifact_id(persisted))
760        .bind(protocol(persisted))
761        .bind(digest(artifact))
762        .bind(persisted)
763        .execute(&mut *transaction)
764        .await?;
765    }
766    transaction.commit().await?;
767    Ok(())
768}
769
770fn persisted_delivery_artifact(artifact: &Value) -> Result<Value, sqlx::Error> {
771    let protocol = protocol(artifact);
772    if !artifact.is_object() {
773        return Err(sqlx::Error::Protocol(
774            "delivery artifact must be an identified Lenso protocol object".to_owned(),
775        ));
776    }
777    let persisted = match protocol {
778        SERVICE_RELEASE_PROTOCOL => validated_canonical_artifact::<ServiceRelease>(
779            artifact,
780            "Service Release",
781            service_release_integrity_is_valid,
782        )?,
783        RELEASE_TRUST_EVIDENCE_PROTOCOL => {
784            canonical_artifact::<ReleaseTrustEvidence>(artifact, "Release Trust Evidence")?
785        }
786        POLICY_EVIDENCE_PROTOCOL => validated_canonical_artifact::<PolicyEvidence>(
787            artifact,
788            "Policy Evidence",
789            policy_evidence_integrity_is_valid,
790        )?,
791        CONFIG_REVISION_PROTOCOL => {
792            let revision = canonical_typed_artifact::<ConfigRevision>(artifact, "Config Revision")?;
793            if !config_revision_integrity_is_valid(&revision)
794                || !revision
795                    .secret_references
796                    .iter()
797                    .all(secret_reference_metadata_is_safe)
798            {
799                return Err(sqlx::Error::Protocol(
800                    "Config Revision identity, digest, or Secret Reference metadata is invalid"
801                        .to_owned(),
802                ));
803            }
804            let mut value = serde_json::to_value(revision).expect("Config Revision serializes");
805            let object = value.as_object_mut().expect("Config Revision is an object");
806            object.insert("values".to_owned(), Value::Object(serde_json::Map::new()));
807            object.insert("valuesRedacted".to_owned(), Value::Bool(true));
808            value
809        }
810        CONFIG_ACTIVATION_RECEIPT_PROTOCOL => {
811            canonical_artifact::<ConfigActivationReceipt>(artifact, "Config Activation Receipt")?
812        }
813        EDGE_CONTRACT_PROTOCOL => validated_canonical_artifact::<EdgeContract>(
814            artifact,
815            "Edge Contract",
816            edge_contract_integrity_is_valid,
817        )?,
818        GATEWAY_PLAN_PROTOCOL => validated_canonical_artifact::<GatewayConfigurationPlan>(
819            artifact,
820            "Gateway Configuration Plan",
821            gateway_plan_integrity_is_valid,
822        )?,
823        GATEWAY_OBSERVATION_PROTOCOL => validated_canonical_artifact::<GatewayObservation>(
824            artifact,
825            "Gateway Observation",
826            gateway_observation_content_integrity_is_valid,
827        )?,
828        DEPLOYMENT_PLAN_PROTOCOL => validated_canonical_artifact::<DeploymentPlan>(
829            artifact,
830            "Deployment Plan",
831            deployment_plan_integrity_is_valid,
832        )?,
833        DEPLOYMENT_RECEIPT_PROTOCOL => {
834            canonical_artifact::<DeploymentReceipt>(artifact, "Deployment Receipt")?
835        }
836        DEPLOYMENT_OBSERVATION_PROTOCOL => validated_canonical_artifact::<DeploymentObservation>(
837            artifact,
838            "Deployment Observation",
839            deployment_observation_content_integrity_is_valid,
840        )?,
841        ENVIRONMENT_VERIFICATION_PROTOCOL => {
842            validated_canonical_artifact::<EnvironmentVerification>(
843                artifact,
844                "Environment Verification",
845                environment_verification_integrity_is_valid,
846            )?
847        }
848        PROMOTION_PLAN_PROTOCOL => validated_canonical_artifact::<PromotionPlan>(
849            artifact,
850            "Promotion Plan",
851            promotion_plan_integrity_is_valid,
852        )?,
853        PROMOTION_APPROVAL_PROTOCOL => {
854            canonical_artifact::<PromotionApproval>(artifact, "Promotion Approval")?
855        }
856        PROMOTION_RECEIPT_PROTOCOL => {
857            canonical_artifact::<PromotionReceipt>(artifact, "Promotion Receipt")?
858        }
859        CANARY_PLAN_PROTOCOL => validated_canonical_artifact::<CanaryPlan>(
860            artifact,
861            "Canary Plan",
862            canary_plan_integrity_is_valid,
863        )?,
864        RELIABILITY_OBSERVATION_PROTOCOL => {
865            canonical_artifact::<ReliabilityObservation>(artifact, "Reliability Observation")?
866        }
867        CANARY_DECISION_PROTOCOL => validated_canonical_artifact::<CanaryDecision>(
868            artifact,
869            "Canary Decision",
870            canary_decision_integrity_is_valid,
871        )?,
872        ROLLBACK_PLAN_PROTOCOL => validated_canonical_artifact::<RollbackPlan>(
873            artifact,
874            "Rollback Plan",
875            rollback_plan_integrity_is_valid,
876        )?,
877        ROLLBACK_RECEIPT_PROTOCOL => {
878            canonical_artifact::<RollbackReceipt>(artifact, "Rollback Receipt")?
879        }
880        DELIVERY_FAILURE_RECOVERY_PROTOCOL => {
881            validated_canonical_artifact::<DeliveryFailureRecoveryEvidence>(
882                artifact,
883                "Delivery Failure Recovery Evidence",
884                delivery_failure_recovery_integrity_is_valid,
885            )?
886        }
887        SERVICE_RESTORE_EVIDENCE_PROTOCOL => {
888            validated_canonical_artifact::<ServiceRestoreEvidence>(
889                artifact,
890                "Service Restore Evidence",
891                service_restore_integrity_is_valid,
892            )?
893        }
894        DISASTER_RECOVERY_EVIDENCE_PROTOCOL => {
895            validated_canonical_artifact::<DisasterRecoveryEvidence>(
896                artifact,
897                "Disaster Recovery Evidence",
898                disaster_recovery_evidence_integrity_is_valid,
899            )?
900        }
901        PERFORMANCE_PROFILE_PROTOCOL => validated_canonical_artifact::<PerformanceProfile>(
902            artifact,
903            "Performance Profile",
904            performance_profile_integrity_is_valid,
905        )?,
906        SUPPORT_ENVELOPE_PROTOCOL => validated_canonical_artifact::<SupportEnvelope>(
907            artifact,
908            "Support Envelope",
909            support_envelope_integrity_is_valid,
910        )?,
911        SECURITY_REVIEW_PROTOCOL => validated_canonical_artifact::<SecurityReviewEvidence>(
912            artifact,
913            "Security Review Evidence",
914            security_review_integrity_is_valid,
915        )?,
916        GA_SUPPORT_MANIFEST_PROTOCOL => validated_canonical_artifact::<GaSupportManifest>(
917            artifact,
918            "GA Support Manifest",
919            ga_support_manifest_integrity_valid,
920        )?,
921        CONTRACT_RETIREMENT_PLAN_PROTOCOL => {
922            validated_canonical_artifact::<ContractRetirementPlan>(
923                artifact,
924                "Contract Retirement Plan",
925                contract_retirement_plan_integrity_is_valid,
926            )?
927        }
928        CONTRACT_RETIREMENT_RECEIPT_PROTOCOL => {
929            validated_canonical_artifact::<ContractRetirementReceipt>(
930                artifact,
931                "Contract Retirement Receipt",
932                contract_retirement_receipt_integrity_is_valid,
933            )?
934        }
935        _ => {
936            return Err(sqlx::Error::Protocol(format!(
937                "unsupported production delivery artifact protocol `{protocol}`"
938            )));
939        }
940    };
941    if delivery_artifact_contains_secret_shaped_field(&persisted) {
942        return Err(sqlx::Error::Protocol(
943            "delivery artifact contains a forbidden secret-shaped field".to_owned(),
944        ));
945    }
946    Ok(persisted)
947}
948
949fn canonical_typed_artifact<T: DeserializeOwned + Serialize>(
950    artifact: &Value,
951    label: &str,
952) -> Result<T, sqlx::Error> {
953    let typed = serde_json::from_value::<T>(artifact.clone())
954        .map_err(|error| sqlx::Error::Protocol(format!("invalid {label} artifact: {error}")))?;
955    let canonical = serde_json::to_value(&typed).expect("typed delivery artifact serializes");
956    if canonical != *artifact {
957        return Err(sqlx::Error::Protocol(format!(
958            "{label} artifact is non-canonical or contains unknown fields"
959        )));
960    }
961    Ok(typed)
962}
963
964fn canonical_artifact<T: DeserializeOwned + Serialize>(
965    artifact: &Value,
966    label: &str,
967) -> Result<Value, sqlx::Error> {
968    canonical_typed_artifact::<T>(artifact, label)?;
969    Ok(artifact.clone())
970}
971
972fn validated_canonical_artifact<T: DeserializeOwned + Serialize>(
973    artifact: &Value,
974    label: &str,
975    is_valid: impl FnOnce(&T) -> bool,
976) -> Result<Value, sqlx::Error> {
977    let typed = canonical_typed_artifact::<T>(artifact, label)?;
978    if !is_valid(&typed) {
979        return Err(sqlx::Error::Protocol(format!(
980            "{label} identity or content digest is invalid"
981        )));
982    }
983    Ok(artifact.clone())
984}
985
986fn delivery_artifact_contains_secret_shaped_field(value: &Value) -> bool {
987    match value {
988        Value::Object(object) => object.iter().any(|(key, value)| {
989            let key = key.to_ascii_lowercase();
990            [
991                "secretvalue",
992                "passwordvalue",
993                "credential",
994                "privatekey",
995                "signingkey",
996                "accesstoken",
997                "token",
998            ]
999            .iter()
1000            .any(|forbidden| key.contains(forbidden))
1001                || delivery_artifact_contains_secret_shaped_field(value)
1002        }),
1003        Value::Array(values) => values
1004            .iter()
1005            .any(delivery_artifact_contains_secret_shaped_field),
1006        _ => false,
1007    }
1008}
1009
1010/// Load the newest persisted delivery evidence and evaluate its read-only projection.
1011pub async fn load_delivery_console_projection(
1012    pool: &sqlx::PgPool,
1013    requested_delivery_id: Option<&str>,
1014) -> Result<DeliveryConsoleProjection, sqlx::Error> {
1015    let exists = sqlx::query_scalar::<_, Option<String>>(
1016        "select to_regclass('platform.delivery_artifacts')::text",
1017    )
1018    .fetch_one(pool)
1019    .await?
1020    .is_some();
1021    if !exists {
1022        return Ok(project_delivery_console(DeliveryConsoleArtifacts {
1023            artifacts: Vec::new(),
1024        }));
1025    }
1026    let delivery_id = match requested_delivery_id {
1027        Some(delivery_id) => Some(delivery_id.to_owned()),
1028        None => sqlx::query_scalar::<_, String>(
1029            "select delivery_id from platform.delivery_artifacts order by recorded_at desc, record_index desc, delivery_id desc limit 1",
1030        )
1031        .fetch_optional(pool)
1032        .await?,
1033    };
1034    let Some(delivery_id) = delivery_id else {
1035        return Ok(project_delivery_console(DeliveryConsoleArtifacts {
1036            artifacts: Vec::new(),
1037        }));
1038    };
1039    let artifacts = sqlx::query_scalar::<_, Value>(
1040        "select artifact_json from platform.delivery_artifacts where delivery_id = $1 order by record_index",
1041    )
1042    .bind(delivery_id)
1043    .fetch_all(pool)
1044    .await?;
1045    Ok(project_delivery_console(DeliveryConsoleArtifacts {
1046        artifacts,
1047    }))
1048}
1049
1050fn derive_state(
1051    artifacts: &[Value],
1052    blocked: bool,
1053    deployments: &[DeliveryConsoleDeployment],
1054) -> DeliveryConsoleState {
1055    if let Some(explicit) = latest(artifacts, "lenso.delivery-state.v1")
1056        .and_then(|artifact| text(artifact, "state"))
1057        .and_then(|state| serde_json::from_value(Value::String(state)).ok())
1058    {
1059        return explicit;
1060    }
1061    let converged = !deployments.is_empty()
1062        && deployments
1063            .iter()
1064            .all(|deployment| deployment.fresh && !deployment.drifted);
1065    let latest_lifecycle = artifacts.iter().rev().find(|artifact| {
1066        matches!(
1067            protocol(artifact),
1068            "lenso.rollback-receipt.v1"
1069                | "lenso.rollback-plan.v1"
1070                | "lenso.canary-decision.v1"
1071                | "lenso.promotion-receipt.v1"
1072                | "lenso.promotion-approval.v1"
1073                | "lenso.environment-verification.v1"
1074        )
1075    });
1076    if let Some(artifact) = latest_lifecycle {
1077        match protocol(artifact) {
1078            "lenso.rollback-receipt.v1" => {
1079                if text(artifact, "outcome").as_deref() == Some("intervention_required") {
1080                    DeliveryConsoleState::InterventionRequired
1081                } else {
1082                    DeliveryConsoleState::RolledBack
1083                }
1084            }
1085            "lenso.rollback-plan.v1" if !boolean(artifact, "automaticAllowed") => {
1086                DeliveryConsoleState::Paused
1087            }
1088            "lenso.rollback-plan.v1" => DeliveryConsoleState::RollingBack,
1089            "lenso.canary-decision.v1" => match text(artifact, "outcome").as_deref() {
1090                Some("rollback") => DeliveryConsoleState::RollingBack,
1091                Some("pause") => DeliveryConsoleState::Paused,
1092                Some("expand" | "hold_degraded") => DeliveryConsoleState::Canary,
1093                Some("converged") => DeliveryConsoleState::Ready,
1094                _ => DeliveryConsoleState::Blocked,
1095            },
1096            "lenso.promotion-receipt.v1" => {
1097                if converged {
1098                    DeliveryConsoleState::Ready
1099                } else {
1100                    DeliveryConsoleState::Converging
1101                }
1102            }
1103            "lenso.promotion-approval.v1" => DeliveryConsoleState::Approved,
1104            "lenso.environment-verification.v1"
1105                if text(artifact, "decision").as_deref() == Some("passed") =>
1106            {
1107                DeliveryConsoleState::Staged
1108            }
1109            _ => DeliveryConsoleState::Blocked,
1110        }
1111    } else if converged {
1112        DeliveryConsoleState::Ready
1113    } else if blocked {
1114        DeliveryConsoleState::Blocked
1115    } else {
1116        DeliveryConsoleState::Planned
1117    }
1118}
1119
1120fn issues_from(artifact: &Value) -> Vec<DeliveryConsoleIssue> {
1121    ["issues", "remainingRisks"]
1122        .into_iter()
1123        .flat_map(|field| array(artifact, field))
1124        .map(|value| DeliveryConsoleIssue {
1125            code: text(value, "code").unwrap_or_else(|| "unknown".to_owned()),
1126            message: text(value, "message").unwrap_or_else(|| "Unknown delivery issue.".to_owned()),
1127            evidence_references: strings(value, "evidenceReferences"),
1128            remediation: text(value, "remediation")
1129                .unwrap_or_else(|| "Inspect the linked delivery evidence.".to_owned()),
1130            next_actions: strings(value, "nextActions"),
1131        })
1132        .collect()
1133}
1134
1135fn timeline_entry(artifact: &Value) -> DeliveryConsoleTimelineEntry {
1136    let protocol = protocol(artifact).to_owned();
1137    let state = text(artifact, "outcome")
1138        .or_else(|| text(artifact, "decision"))
1139        .or_else(|| text(artifact, "activation"))
1140        .unwrap_or_else(|| "recorded".to_owned());
1141    let mut evidence_references = strings(artifact, "evidenceReferences");
1142    evidence_references.extend(
1143        issues_from(artifact)
1144            .into_iter()
1145            .flat_map(|issue| issue.evidence_references),
1146    );
1147    evidence_references.sort();
1148    evidence_references.dedup();
1149    DeliveryConsoleTimelineEntry {
1150        protocol,
1151        artifact_id: artifact_id(artifact),
1152        state,
1153        evidence_references,
1154    }
1155}
1156
1157fn filter_timeline(
1158    timeline: &[DeliveryConsoleTimelineEntry],
1159    needle: &str,
1160) -> Vec<DeliveryConsoleTimelineEntry> {
1161    timeline
1162        .iter()
1163        .filter(|entry| entry.protocol.contains(needle))
1164        .cloned()
1165        .collect()
1166}
1167
1168fn is_timeline_protocol(protocol: &str) -> bool {
1169    ["promotion", "canary", "rollback", "config-activation"]
1170        .iter()
1171        .any(|needle| protocol.contains(needle))
1172}
1173
1174fn collect_runtime_story_references(value: &Value, found: &mut BTreeSet<String>) {
1175    match value {
1176        Value::String(value) if value.starts_with("runtime-story:") => {
1177            found.insert(value.clone());
1178        }
1179        Value::Array(values) => {
1180            for value in values {
1181                collect_runtime_story_references(value, found);
1182            }
1183        }
1184        Value::Object(values) => {
1185            for value in values.values() {
1186                collect_runtime_story_references(value, found);
1187            }
1188        }
1189        _ => {}
1190    }
1191}
1192
1193fn latest<'a>(artifacts: &'a [Value], expected_protocol: &str) -> Option<&'a Value> {
1194    artifacts
1195        .iter()
1196        .rev()
1197        .find(|artifact| protocol(artifact) == expected_protocol)
1198}
1199
1200fn protocol(value: &Value) -> &str {
1201    value
1202        .get("protocol")
1203        .and_then(Value::as_str)
1204        .unwrap_or("unknown")
1205}
1206
1207fn artifact_id(value: &Value) -> String {
1208    [
1209        "releaseId",
1210        "receiptId",
1211        "decisionId",
1212        "evidenceId",
1213        "verificationId",
1214        "planId",
1215        "revisionId",
1216        "contractId",
1217        "observationId",
1218        "proofId",
1219    ]
1220    .into_iter()
1221    .find_map(|field| text(value, field))
1222    .unwrap_or_else(|| format!("artifact:{}", digest(value)))
1223}
1224
1225fn text(value: &Value, field: &str) -> Option<String> {
1226    value.get(field).and_then(Value::as_str).map(str::to_owned)
1227}
1228
1229fn nested_text(value: &Value, parent: &str, field: &str) -> Option<String> {
1230    value.get(parent).and_then(|value| text(value, field))
1231}
1232
1233fn boolean(value: &Value, field: &str) -> bool {
1234    value.get(field).and_then(Value::as_bool).unwrap_or(false)
1235}
1236
1237fn array<'a>(value: &'a Value, field: &str) -> Vec<&'a Value> {
1238    value
1239        .get(field)
1240        .and_then(Value::as_array)
1241        .map(|values| values.iter().collect())
1242        .unwrap_or_default()
1243}
1244
1245fn strings(value: &Value, field: &str) -> Vec<String> {
1246    array(value, field)
1247        .into_iter()
1248        .filter_map(Value::as_str)
1249        .map(str::to_owned)
1250        .collect()
1251}
1252
1253fn digest(value: &impl Serialize) -> String {
1254    extraction_input_digest(
1255        serde_json::to_vec(value).expect("delivery projection values serialize"),
1256    )
1257}
1258
1259#[cfg(test)]
1260mod persistence_tests {
1261    use super::*;
1262    use std::collections::BTreeMap;
1263
1264    #[test]
1265    fn persistence_rejects_unknown_delivery_protocols() {
1266        let error = persisted_delivery_artifact(&serde_json::json!({
1267            "protocol": "lenso.forged-delivery-object.v1",
1268            "receiptId": "forged:1"
1269        }))
1270        .expect_err("unknown delivery objects must fail closed");
1271
1272        assert!(error.to_string().contains("unsupported"));
1273    }
1274
1275    #[test]
1276    fn persistence_rejects_extra_token_fields_on_typed_receipts() {
1277        let receipt = crate::RollbackReceipt {
1278            protocol: crate::ROLLBACK_RECEIPT_PROTOCOL.to_owned(),
1279            receipt_id: "rollback-receipt:test".to_owned(),
1280            plan_id: "rollback-plan:test".to_owned(),
1281            actor: "automation:test".to_owned(),
1282            outcome: crate::RollbackOutcome::RolledBack,
1283            restored_release_id: "release:previous".to_owned(),
1284            restored_config_revision_id: "config:previous".to_owned(),
1285            environment_revision_before: 7,
1286            environment_revision_after: 8,
1287            exposure_percent: 0,
1288            remaining_risks: Vec::new(),
1289            approval_boundary_required: false,
1290            evidence_references: Vec::new(),
1291            effects: crate::DeliveryEffects {
1292                mutates_environment: true,
1293                mutates_configuration: true,
1294                mutates_gateway: true,
1295                mutates_deployment: true,
1296                appends_ledger: true,
1297            },
1298        };
1299        let mut artifact = serde_json::to_value(receipt).expect("receipt serializes");
1300        artifact
1301            .as_object_mut()
1302            .expect("receipt is an object")
1303            .insert(
1304                "metadata".to_owned(),
1305                serde_json::json!({"token": "forged"}),
1306            );
1307
1308        let error = persisted_delivery_artifact(&artifact)
1309            .expect_err("unknown token-bearing receipt fields must fail closed");
1310        assert!(error.to_string().contains("non-canonical"));
1311    }
1312
1313    #[test]
1314    fn persistence_accepts_integrity_valid_ga_support_and_rejects_tampering() {
1315        let manifest = crate::assemble_ga_support_manifest_with_trust(
1316            crate::GaSupportManifestInput {
1317                status: crate::SupportStatus::Candidate,
1318                components: vec![crate::GaComponent {
1319                    kind: crate::ComponentKind::Runtime,
1320                    component_id: "lenso-service".to_owned(),
1321                    version: "0.1.14".to_owned(),
1322                    digest: crate::extraction_input_digest(b"runtime"),
1323                }],
1324                manifest_formats: vec![crate::ManifestFormat {
1325                    kind: crate::ManifestKind::Service,
1326                    version: "lenso.service.v2".to_owned(),
1327                }],
1328                state_versions: vec!["service-store.v1".to_owned()],
1329                adapter_versions: BTreeMap::from([("postgresql".to_owned(), "18".to_owned())]),
1330                documentation: crate::DocumentationIdentity {
1331                    version: "m6-ga".to_owned(),
1332                    digest: crate::extraction_input_digest(b"docs"),
1333                },
1334                combinations: vec![crate::SupportCombinationInput {
1335                    combination_id: "candidate".to_owned(),
1336                    component_references: vec!["runtime:lenso-service@0.1.14".to_owned()],
1337                    state_version: "service-store.v1".to_owned(),
1338                    status: crate::SupportStatus::Candidate,
1339                }],
1340                upgrade_edges: Vec::new(),
1341            },
1342            crate::EvidenceReceiptTrust {
1343                authorities: BTreeMap::from([(
1344                    crate::PERFORMANCE_PROFILE_PROTOCOL.to_owned(),
1345                    "test-authority".to_owned(),
1346                )]),
1347                public_keys: BTreeMap::from([(
1348                    "test-authority".to_owned(),
1349                    "-----BEGIN PUBLIC KEY-----\ntest\n-----END PUBLIC KEY-----".to_owned(),
1350                )]),
1351            },
1352        )
1353        .expect("manifest is valid");
1354        let artifact = serde_json::to_value(&manifest).expect("manifest serializes");
1355        assert_eq!(
1356            persisted_delivery_artifact(&artifact).expect("manifest persists"),
1357            artifact
1358        );
1359
1360        let mut tampered = artifact;
1361        tampered["status"] = serde_json::json!("general_availability");
1362        persisted_delivery_artifact(&tampered).expect_err("tampered manifest fails");
1363    }
1364}
lenso_service/delivery_console_projection.rs

lenso_service/
delivery_console_projection.rs
