synapse_application/
mutation_dispatch.rs

1use serde_json::Value;
2use soma_infra::MutationProgressReporter;
3use soma_ops::{AuthorizationEvidence, OperationContext, OperationName, OperationPlan};
4use tokio_util::sync::CancellationToken;
5
6use crate::mutation_runtime::lifecycle_action;
7use crate::{ExecutionError, SynapseMutationRuntime};
8
9impl SynapseMutationRuntime {
10    /// Builds a deterministic, topology-bound plan for a supported mutation.
11    pub async fn plan(
12        &self,
13        operation: &OperationName,
14        parameters: &Value,
15        context: &OperationContext,
16    ) -> Result<OperationPlan, ExecutionError> {
17        if lifecycle_action(operation).is_ok() {
18            self.plan_container(operation, parameters, context).await
19        } else if crate::mutation_compose::compose_action(operation).is_ok() {
20            self.plan_compose(operation, parameters, context).await
21        } else if crate::mutation_pull::pull_operation(operation) {
22            self.plan_pull(operation, parameters, context).await
23        } else if crate::mutation_build::build_operation(operation) {
24            self.plan_build(operation, parameters, context).await
25        } else if crate::mutation_recreate::recreate_operation(operation) {
26            self.plan_recreate(operation, parameters, context).await
27        } else if crate::mutation_exec::exec_operation(operation) {
28            self.plan_exec(operation, parameters, context).await
29        } else if crate::mutation_final_contract::final_operation(operation) {
30            self.plan_final(operation, parameters, context).await
31        } else {
32            Err(ExecutionError::UnsupportedOperation(operation.clone()))
33        }
34    }
35
36    /// Executes one supported mutation while intentionally discarding progress events.
37    pub async fn execute(
38        &self,
39        operation: &OperationName,
40        parameters: &Value,
41        context: &OperationContext,
42        plan: &OperationPlan,
43        authorization: &AuthorizationEvidence,
44        cancellation: &CancellationToken,
45    ) -> Result<soma_ops::OperationResult, ExecutionError> {
46        self.execute_with_progress(
47            operation,
48            parameters,
49            context,
50            plan,
51            authorization,
52            &soma_ops::NoopProgressSink,
53            cancellation,
54        )
55        .await
56    }
57
58    /// Executes one supported mutation with canonical progress delivery.
59    #[allow(clippy::too_many_arguments)]
60    pub async fn execute_with_progress(
61        &self,
62        operation: &OperationName,
63        parameters: &Value,
64        context: &OperationContext,
65        plan: &OperationPlan,
66        authorization: &AuthorizationEvidence,
67        progress: &dyn MutationProgressReporter,
68        cancellation: &CancellationToken,
69    ) -> Result<soma_ops::OperationResult, ExecutionError> {
70        if lifecycle_action(operation).is_ok() {
71            self.execute_container(
72                operation,
73                parameters,
74                context,
75                plan,
76                authorization,
77                cancellation,
78            )
79            .await
80        } else if crate::mutation_compose::compose_action(operation).is_ok() {
81            self.execute_compose(
82                operation,
83                parameters,
84                context,
85                plan,
86                authorization,
87                cancellation,
88            )
89            .await
90        } else if crate::mutation_pull::pull_operation(operation) {
91            self.execute_pull(
92                operation,
93                parameters,
94                context,
95                plan,
96                authorization,
97                progress,
98                cancellation,
99            )
100            .await
101        } else if crate::mutation_build::build_operation(operation) {
102            self.execute_build(
103                operation,
104                parameters,
105                context,
106                plan,
107                authorization,
108                progress,
109                cancellation,
110            )
111            .await
112        } else if crate::mutation_recreate::recreate_operation(operation) {
113            self.execute_recreate(
114                operation,
115                parameters,
116                context,
117                plan,
118                authorization,
119                cancellation,
120            )
121            .await
122        } else if crate::mutation_exec::exec_operation(operation) {
123            self.execute_exec(
124                operation,
125                parameters,
126                context,
127                plan,
128                authorization,
129                cancellation,
130            )
131            .await
132        } else if crate::mutation_final_contract::final_operation(operation) {
133            self.execute_final(
134                operation,
135                parameters,
136                context,
137                plan,
138                authorization,
139                cancellation,
140            )
141            .await
142        } else {
143            Err(ExecutionError::UnsupportedOperation(operation.clone()))
144        }
145    }
146}
147
148#[cfg(test)]
149#[path = "mutation_dispatch_tests.rs"]
150mod tests;
synapse_application/mutation_dispatch.rs

synapse_application/
mutation_dispatch.rs
