appcore_core/
controller.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: controller.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/05/29 20:47:35 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/07/23 23:50:45 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Minimal runtime controller for lifecycle mutation and command dispatch delegation.
12
13use crate::audit::{AuditCategory, AuditEntry, AuditOutcome, AuditRecord};
14use crate::context::RuntimeContext;
15use crate::envelope::CommandEnvelope;
16use crate::error::{RuntimeError, RuntimeResult};
17use crate::handler::CommandResult;
18use crate::idempotency::{
19    IdempotencyRecord, IdempotencyStatus, IdempotencyStore, InMemoryIdempotencyStore,
20};
21use crate::lifecycle::{RuntimeLifecycle, RuntimeLifecycleEvent, RuntimeLifecycleState};
22use crate::runtime::RuntimeInstance;
23use parking_lot::Mutex;
24use std::sync::Arc;
25
26/// Coordinates lifecycle, idempotency, command dispatch, events and audit.
27pub struct RuntimeController {
28    instance: Arc<RuntimeInstance>,
29    idempotency: Mutex<Box<dyn IdempotencyStore + Send + Sync>>,
30}
31
32impl RuntimeController {
33    /// Creates a controller with process-local idempotency.
34    pub fn new(instance: RuntimeInstance) -> Self {
35        Self {
36            instance: Arc::new(instance),
37            idempotency: Mutex::new(Box::new(InMemoryIdempotencyStore::new())),
38        }
39    }
40
41    /// Creates a controller with an explicit idempotency store.
42    pub fn with_idempotency_store(
43        instance: RuntimeInstance,
44        idempotency: Box<dyn IdempotencyStore + Send + Sync>,
45    ) -> Self {
46        Self {
47            instance: Arc::new(instance),
48            idempotency: Mutex::new(idempotency),
49        }
50    }
51
52    /// Returns the hosted immutable Runtime instance.
53    pub fn instance(&self) -> &RuntimeInstance {
54        &self.instance
55    }
56
57    /// Returns a shared reference-counted Runtime instance.
58    pub fn instance_arc(&self) -> Arc<RuntimeInstance> {
59        self.instance.clone()
60    }
61
62    /// Returns the process lifecycle.
63    pub fn lifecycle(&self) -> &RuntimeLifecycle {
64        self.instance.lifecycle()
65    }
66
67    /// Returns the number of active idempotency records.
68    pub fn idempotency_len(&self) -> usize {
69        self.idempotency.lock().len()
70    }
71
72    /// Reports whether an idempotency key has an active record.
73    pub fn idempotency_contains(&self, key: &str) -> RuntimeResult<bool> {
74        Ok(self.idempotency.lock().get(key)?.is_some())
75    }
76
77    /// Applies one process lifecycle event.
78    pub fn apply_lifecycle_event(
79        &mut self,
80        event: RuntimeLifecycleEvent,
81    ) -> RuntimeResult<RuntimeLifecycleState> {
82        self.instance.lifecycle().apply(event)
83    }
84
85    /// Runs the complete command dispatch transaction.
86    pub fn dispatch_command(
87        &mut self,
88        command: &CommandEnvelope,
89        context: &dyn RuntimeContext,
90    ) -> RuntimeResult<CommandResult> {
91        match self.pre_dispatch(command)? {
92            Some(Ok(replay)) => Ok(replay),
93            Some(Err(err)) => Err(err),
94            None => {
95                let result = self.instance.dispatch_command(command, context);
96                self.post_dispatch(command, &result)?;
97                result
98            }
99        }
100    }
101
102    /// Performs lifecycle and idempotency checks before handler execution.
103    pub fn pre_dispatch(
104        &self,
105        command: &CommandEnvelope,
106    ) -> RuntimeResult<Option<RuntimeResult<CommandResult>>> {
107        if let Some(rejection) = self.check_lifecycle_readiness() {
108            let res = Ok(rejection);
109            self.record_audit(command, &res);
110            return Ok(Some(res));
111        }
112
113        if let Some(key) = command.idempotency_key.as_deref() {
114            let mut store = self.idempotency.lock();
115            if let Some(record) = store.get(key)? {
116                let payload_hash = hash_payload(&command.payload);
117                if record.request_hash != payload_hash {
118                    let err = RuntimeError::IdempotencyConflict {
119                        key: key.to_string(),
120                    };
121                    self.record_audit(command, &Err(err.clone()));
122                    return Err(err);
123                }
124                match record.status {
125                    IdempotencyStatus::Pending => {
126                        let err = RuntimeError::IdempotencyPending {
127                            key: key.to_string(),
128                        };
129                        self.record_audit(command, &Err(err.clone()));
130                        return Err(err);
131                    }
132                    IdempotencyStatus::Resolved {
133                        response_status,
134                        ref response_body,
135                    } => {
136                        if response_status >= 400 {
137                            store.remove(key)?;
138                            return Ok(None);
139                        }
140                        let result = if response_body.is_empty() {
141                            CommandResult::accepted(Vec::new())
142                        } else {
143                            serde_json::from_str::<CommandResult>(response_body).map_err(|e| {
144                                RuntimeError::IdempotencyStoreIo {
145                                    operation: "deserialize_replay",
146                                    message: e.to_string(),
147                                }
148                            })?
149                        };
150                        let res = Ok(result);
151                        self.record_audit(command, &res);
152                        return Ok(Some(res));
153                    }
154                }
155            }
156
157            // Insert Pending record
158            let payload_hash = hash_payload(&command.payload);
159            let record = IdempotencyRecord {
160                key: key.to_string(),
161                request_hash: payload_hash,
162                status: IdempotencyStatus::Pending,
163                created_at_ms: now_ms(),
164            };
165            store.insert(record)?;
166        }
167
168        Ok(None)
169    }
170
171    /// Persists command outcome and emits accepted events after handler execution.
172    pub fn post_dispatch(
173        &self,
174        command: &CommandEnvelope,
175        dispatch_result: &RuntimeResult<CommandResult>,
176    ) -> RuntimeResult<()> {
177        if let Some(key) = command.idempotency_key.as_ref() {
178            let mut store = self.idempotency.lock();
179            match dispatch_result {
180                Ok(result) => {
181                    let response_body = serde_json::to_string(result).map_err(|e| {
182                        RuntimeError::IdempotencyStoreIo {
183                            operation: "serialize_response",
184                            message: e.to_string(),
185                        }
186                    })?;
187                    let record = IdempotencyRecord {
188                        key: key.to_string(),
189                        request_hash: hash_payload(&command.payload),
190                        status: IdempotencyStatus::Resolved {
191                            response_status: 200,
192                            response_body,
193                        },
194                        created_at_ms: now_ms(),
195                    };
196                    store.insert(record)?;
197                }
198                Err(_) => {
199                    store.remove(key)?;
200                }
201            }
202        }
203
204        self.emit_events_and_audit(command, dispatch_result);
205        Ok(())
206    }
207
208    fn check_lifecycle_readiness(&self) -> Option<CommandResult> {
209        match self.lifecycle().current() {
210            RuntimeLifecycleState::Running | RuntimeLifecycleState::Degraded => None,
211            RuntimeLifecycleState::Restricted => {
212                Some(CommandResult::rejected("runtime is restricted"))
213            }
214            _ => Some(CommandResult::rejected("runtime is not ready")),
215        }
216    }
217
218    fn emit_events_and_audit(
219        &self,
220        command: &CommandEnvelope,
221        dispatch_result: &RuntimeResult<CommandResult>,
222    ) {
223        if let Ok(result) = dispatch_result {
224            if result.is_accepted() {
225                let events = result
226                    .events()
227                    .iter()
228                    .cloned()
229                    .map(|event| {
230                        if event.trace.is_none() {
231                            if let Some(trace) = &command.trace {
232                                return event.with_trace(trace.clone());
233                            }
234                        }
235                        event
236                    })
237                    .collect::<Vec<_>>();
238                for event in &events {
239                    let completed_at_ms = now_ms();
240                    self.instance.audit_log().push_entry(
241                        AuditEntry::new(
242                            AuditCategory::Event,
243                            event.event_id.clone(),
244                            event.event_name.as_str(),
245                            event.occurred_at_ms,
246                            completed_at_ms,
247                            AuditOutcome::Accepted,
248                        )
249                        .with_runtime_scope(&event.app_id, &event.node_id)
250                        .with_trace(event.trace.clone()),
251                    );
252                }
253                self.instance.event_bus().emit_many(events);
254            }
255        }
256        self.record_audit(command, dispatch_result);
257    }
258
259    fn record_audit(
260        &self,
261        command: &CommandEnvelope,
262        dispatch_result: &RuntimeResult<CommandResult>,
263    ) {
264        let record = match dispatch_result {
265            Ok(result) => AuditRecord {
266                command_id: command.command_id.clone(),
267                command_name: command.command_name.clone(),
268                app_id: command.app_id.clone(),
269                node_id: command.node_id.clone(),
270                timestamp_ms: command.issued_at_ms,
271                outcome: if result.is_accepted() {
272                    AuditOutcome::Accepted
273                } else {
274                    AuditOutcome::Rejected
275                },
276                message: result.message().map(|msg| msg.to_string()),
277                trace: command.trace.clone(),
278            },
279            Err(error) => AuditRecord {
280                command_id: command.command_id.clone(),
281                command_name: command.command_name.clone(),
282                app_id: command.app_id.clone(),
283                node_id: command.node_id.clone(),
284                timestamp_ms: command.issued_at_ms,
285                outcome: AuditOutcome::Error,
286                message: Some(format!("{error:?}")),
287                trace: command.trace.clone(),
288            },
289        };
290
291        self.instance.audit_log().push(record);
292    }
293}
294
295fn hash_payload(payload: &[u8]) -> String {
296    use sha2::{Digest, Sha256};
297    let mut hasher = Sha256::new();
298    hasher.update(payload);
299    let result = hasher.finalize();
300    let mut hex = String::with_capacity(result.len() * 2);
301    for byte in result {
302        hex.push_str(&format!("{:02x}", byte));
303    }
304    hex
305}
306
307fn now_ms() -> u64 {
308    std::time::SystemTime::now()
309        .duration_since(std::time::UNIX_EPOCH)
310        .map(|d| d.as_millis() as u64)
311        .unwrap_or(0)
312}
313
314#[cfg(test)]
315mod controller_tests;
appcore_core/controller.rs

appcore_core/
controller.rs
