made_app/workers/
execute_ceremony_operation_use_case.rs

1use std::sync::Arc;
2
3use made_core::error::DomainError;
4use made_core::ports::{
5    CeremonyExecutionConnectorOutcome, CeremonyExecutionConnectorPort, CeremonyExecutionRequest,
6    ClockPort, ExecutionReceiptStorePort, RecordExecutionIntentOutcome,
7};
8use made_core::value_objects::{ExecutionIntent, ExecutionOperation, ExecutionRecoveryCapability};
9
10use super::execution_receipt_artifact_verifier::verify_receipt_artifacts;
11use super::execution_receipt_from_observation::execution_receipt_from_observation;
12use super::{ExecuteCeremonyOperationInput, ExecuteCeremonyOperationOutcome};
13use crate::artifacts::ArtifactService;
14
15/// Persist intent, execute or recover once, then persist an immutable receipt.
16pub struct ExecuteCeremonyOperationUseCase {
17    store: Arc<dyn ExecutionReceiptStorePort>,
18    connector: Arc<dyn CeremonyExecutionConnectorPort>,
19    clock: Arc<dyn ClockPort>,
20    artifacts: Option<Arc<ArtifactService>>,
21}
22
23impl std::fmt::Debug for ExecuteCeremonyOperationUseCase {
24    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
25        formatter
26            .debug_struct("ExecuteCeremonyOperationUseCase")
27            .field("connector_id", &self.connector.connector_id())
28            .finish_non_exhaustive()
29    }
30}
31
32impl ExecuteCeremonyOperationUseCase {
33    #[must_use]
34    pub fn new(
35        store: Arc<dyn ExecutionReceiptStorePort>,
36        connector: Arc<dyn CeremonyExecutionConnectorPort>,
37        clock: Arc<dyn ClockPort>,
38    ) -> Self {
39        Self {
40            store,
41            connector,
42            clock,
43            artifacts: None,
44        }
45    }
46
47    #[must_use]
48    pub fn with_artifacts(mut self, artifacts: Arc<ArtifactService>) -> Self {
49        self.artifacts = Some(artifacts);
50        self
51    }
52
53    pub async fn execute(
54        &self,
55        input: ExecuteCeremonyOperationInput,
56    ) -> Result<ExecuteCeremonyOperationOutcome, DomainError> {
57        let semantic_request = input.handler_request.semantic_request_bytes()?;
58        let candidate = ExecutionOperation::new(
59            input.handler_request.instance_id().clone(),
60            input.handler_request.step_id().clone(),
61            input.state_visit,
62            input.state_iteration,
63            input.step_iteration,
64            semantic_request,
65        );
66        let operation = match self.store.operation(candidate.operation_id()).await? {
67            Some(stored) if stored == candidate => stored,
68            Some(_) => {
69                return Err(DomainError::Conflict {
70                    what: "execution_operation",
71                })
72            }
73            None => candidate,
74        };
75        let existing_intent = self
76            .store
77            .intent(operation.operation_id(), &input.claim_fence)
78            .await?;
79        let (intent, recorded) = if let Some(intent) = existing_intent {
80            if intent.operation() != &operation
81                || intent.connector_id() != self.connector.connector_id()
82                || intent.recovery_capability() != self.connector.recovery_capability()
83                || intent.source_kind() != self.connector.source_kind()
84                || intent.actor_kind() != input.actor_kind
85            {
86                return Err(DomainError::Conflict {
87                    what: "execution_intent",
88                });
89            }
90            (intent, RecordExecutionIntentOutcome::AlreadyRecorded)
91        } else {
92            let intent = ExecutionIntent::new(
93                operation,
94                input.claim_fence,
95                self.connector.connector_id().clone(),
96                self.connector.recovery_capability(),
97                self.connector.source_kind(),
98                input.actor_kind,
99                self.clock.now(),
100            )?;
101            let recorded = self.store.record_intent(intent.clone()).await?;
102            (intent, recorded)
103        };
104        if let Some(receipt) = self
105            .store
106            .receipt(intent.operation().operation_id())
107            .await?
108        {
109            verify_receipt_artifacts(self.artifacts.as_deref(), &receipt).await?;
110            return Ok(ExecuteCeremonyOperationOutcome::Receipt(Box::new(receipt)));
111        }
112        if recorded != RecordExecutionIntentOutcome::RecordedFirst
113            && self.connector.recovery_capability()
114                == ExecutionRecoveryCapability::ReconciliationRequired
115        {
116            return Ok(ExecuteCeremonyOperationOutcome::ReconciliationRequired(
117                intent.operation().operation_id().clone(),
118            ));
119        }
120
121        let connector_outcome = self
122            .connector
123            .execute_or_recover(CeremonyExecutionRequest::new(
124                intent.clone(),
125                input.handler_request,
126            )?)
127            .await?;
128        let observation = match connector_outcome {
129            CeremonyExecutionConnectorOutcome::Observed(observation) => *observation,
130            CeremonyExecutionConnectorOutcome::ReconciliationRequired(operation_id) => {
131                return Ok(ExecuteCeremonyOperationOutcome::ReconciliationRequired(
132                    operation_id,
133                ));
134            }
135        };
136        let receipt = execution_receipt_from_observation(
137            self.store.as_ref(),
138            self.connector.as_ref(),
139            &intent,
140            observation,
141        )
142        .await?;
143        verify_receipt_artifacts(self.artifacts.as_deref(), &receipt).await?;
144        self.store.record_receipt(receipt.clone()).await?;
145        Ok(ExecuteCeremonyOperationOutcome::Receipt(Box::new(receipt)))
146    }
147}
made_app/workers/execute_ceremony_operation_use_case.rs

made_app/workers/
execute_ceremony_operation_use_case.rs
