mocra_core/engine/events/
events.rs

1use crate::common::model::message::{TaskErrorEvent, TaskEvent, TaskParserEvent};
2use crate::common::model::{Request, Response};
3use crate::engine::task::module::Module;
4use crate::errors::{Error, ErrorKind};
5use serde::{Deserialize, Serialize};
6use uuid::Uuid;
7
8/// Top-level domain that namespaces event families.
9#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)]
10pub enum EventDomain {
11    Engine,
12    System,
13}
14
15/// Canonical event type identifiers emitted by the engine.
16#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)]
17pub enum EventType {
18    TaskModel,
19    ParserTaskModel,
20    ParserTaskProduced,
21    ErrorTaskProduced,
22    ModuleStepAdvanced,
23    ModuleStepFallback,
24    TaskTerminatedByThreshold,
25    RequestPublish,
26    RequestMiddleware,
27    Download,
28    ResponseMiddleware,
29    ResponsePublish,
30    ModuleGenerate,
31    Parser,
32    MiddlewareBefore,
33    DataStore,
34    SystemError,
35    SystemHealth,
36}
37
38/// Lifecycle phase attached to each event emission.
39#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)]
40pub enum EventPhase {
41    Started,
42    Completed,
43    Failed,
44    Retry,
45}
46
47/// Error payload carried by failed events.
48#[derive(Debug, Clone, Serialize, Deserialize)]
49pub struct EventError {
50    pub kind: ErrorKind,
51    pub message: String,
52}
53
54/// Transport envelope used by the event bus and storage backends.
55#[derive(Debug, Clone, Serialize, Deserialize)]
56pub struct EventEnvelope {
57    pub domain: EventDomain,
58    pub event_type: EventType,
59    pub phase: EventPhase,
60    pub payload: serde_json::Value,
61    pub error: Option<EventError>,
62    pub timestamp_ms: u128,
63    pub trace_id: Option<String>,
64}
65
66impl EventDomain {
67    /// Returns the canonical lowercase domain key.
68    ///
69    /// ```
70    /// use mocra_core::engine::events::EventDomain;
71    /// assert_eq!(EventDomain::Engine.as_str(), "engine");
72    /// assert_eq!(EventDomain::System.as_str(), "system");
73    /// ```
74    pub fn as_str(&self) -> &'static str {
75        match self {
76            EventDomain::Engine => "engine",
77            EventDomain::System => "system",
78        }
79    }
80}
81
82impl EventType {
83    /// Returns the canonical lowercase event type key.
84    ///
85    /// ```
86    /// use mocra_core::engine::events::EventType;
87    /// assert_eq!(EventType::RequestPublish.as_str(), "request_publish");
88    /// ```
89    pub fn as_str(&self) -> &'static str {
90        match self {
91            EventType::TaskModel => "task_model",
92            EventType::ParserTaskModel => "parser_task_model",
93            EventType::ParserTaskProduced => "parser_task_produced",
94            EventType::ErrorTaskProduced => "error_task_produced",
95            EventType::ModuleStepAdvanced => "module_step_advanced",
96            EventType::ModuleStepFallback => "module_step_fallback",
97            EventType::TaskTerminatedByThreshold => "task_terminated_by_threshold",
98            EventType::RequestPublish => "request_publish",
99            EventType::RequestMiddleware => "request_middleware",
100            EventType::Download => "download",
101            EventType::ResponseMiddleware => "response_middleware",
102            EventType::ResponsePublish => "response_publish",
103            EventType::ModuleGenerate => "module_generate",
104            EventType::Parser => "parser",
105            EventType::MiddlewareBefore => "middleware_before",
106            EventType::DataStore => "data_store",
107            EventType::SystemError => "system_error",
108            EventType::SystemHealth => "system_health",
109        }
110    }
111}
112
113impl EventPhase {
114    /// Returns the canonical lowercase phase key.
115    ///
116    /// ```
117    /// use mocra_core::engine::events::EventPhase;
118    /// assert_eq!(EventPhase::Retry.as_str(), "retry");
119    /// ```
120    pub fn as_str(&self) -> &'static str {
121        match self {
122            EventPhase::Started => "started",
123            EventPhase::Completed => "completed",
124            EventPhase::Failed => "failed",
125            EventPhase::Retry => "retry",
126        }
127    }
128}
129
130impl EventEnvelope {
131    pub fn engine<T: Serialize>(event_type: EventType, phase: EventPhase, payload: T) -> Self {
132        Self {
133            domain: EventDomain::Engine,
134            event_type,
135            phase,
136            payload: serde_json::to_value(payload).unwrap_or_else(|_| serde_json::json!({})),
137            error: None,
138            timestamp_ms: Self::now_ms(),
139            trace_id: None,
140        }
141    }
142
143    pub fn engine_error<T: Serialize>(
144        event_type: EventType,
145        phase: EventPhase,
146        payload: T,
147        err: &Error,
148    ) -> Self {
149        Self {
150            domain: EventDomain::Engine,
151            event_type,
152            phase,
153            payload: serde_json::to_value(payload).unwrap_or_else(|_| serde_json::json!({})),
154            error: Some(EventError {
155                kind: err.kind().clone(),
156                message: err.to_string(),
157            }),
158            timestamp_ms: Self::now_ms(),
159            trace_id: None,
160        }
161    }
162
163    pub fn system_error(message: impl Into<String>, phase: EventPhase) -> Self {
164        Self {
165            domain: EventDomain::System,
166            event_type: EventType::SystemError,
167            phase,
168            payload: serde_json::json!({ "message": message.into() }),
169            error: None,
170            timestamp_ms: Self::now_ms(),
171            trace_id: None,
172        }
173    }
174
175    pub fn system_health(payload: HealthCheckEvent, phase: EventPhase) -> Self {
176        Self {
177            domain: EventDomain::System,
178            event_type: EventType::SystemHealth,
179            phase,
180            payload: serde_json::to_value(payload).unwrap_or_else(|_| serde_json::json!({})),
181            error: None,
182            timestamp_ms: Self::now_ms(),
183            trace_id: None,
184        }
185    }
186
187    /// Builds the stable composite event key `domain.type.phase`.
188    ///
189    /// ```
190    /// use mocra_core::engine::events::{EventEnvelope, EventPhase, EventType};
191    /// let evt = EventEnvelope::engine(EventType::ParserTaskModel, EventPhase::Completed, serde_json::json!({}));
192    /// assert_eq!(evt.event_key(), "engine.parser_task_model.completed");
193    /// ```
194    pub fn event_key(&self) -> String {
195        format!(
196            "{}.{}.{}",
197            self.domain.as_str(),
198            self.event_type.as_str(),
199            self.phase.as_str()
200        )
201    }
202
203    pub fn now_ms() -> u128 {
204        std::time::SystemTime::now()
205            .duration_since(std::time::UNIX_EPOCH)
206            .unwrap()
207            .as_millis()
208    }
209}
210
211#[cfg(test)]
212mod tests {
213    use super::{EventEnvelope, EventPhase, EventType, ParserTaskModelEvent};
214    use crate::common::model::ExecutionMark;
215    use crate::common::model::message::{TaskErrorEvent, TaskEvent, TaskParserEvent};
216    use serde_json::json;
217    use uuid::Uuid;
218
219    #[test]
220    fn threshold_termination_event_type_has_stable_name() {
221        assert_eq!(
222            EventType::TaskTerminatedByThreshold.as_str(),
223            "task_terminated_by_threshold"
224        );
225    }
226
227    #[test]
228    fn task_model_chain_semantic_event_types_have_stable_names() {
229        assert_eq!(
230            EventType::ParserTaskProduced.as_str(),
231            "parser_task_produced"
232        );
233        assert_eq!(EventType::ErrorTaskProduced.as_str(), "error_task_produced");
234        assert_eq!(
235            EventType::ModuleStepAdvanced.as_str(),
236            "module_step_advanced"
237        );
238        assert_eq!(
239            EventType::ModuleStepFallback.as_str(),
240            "module_step_fallback"
241        );
242    }
243
244    #[test]
245    fn parser_task_model_event_mapping_is_consistent_for_parser_and_error_inputs() {
246        let run_id = Uuid::now_v7();
247        let task_model = TaskEvent {
248            account: "acc".to_string(),
249            platform: "pf".to_string(),
250            module: Some(vec!["m1".to_string()]),
251            priority: Default::default(),
252            run_id,
253        };
254
255        let parser_task = TaskParserEvent {
256            id: Uuid::now_v7(),
257            account_task: task_model.clone(),
258            timestamp: 0,
259            metadata: json!({"k":"v"}).as_object().cloned().unwrap_or_default(),
260            context: ExecutionMark::default(),
261            run_id,
262            prefix_request: Uuid::now_v7(),
263        };
264        let parser_evt = ParserTaskModelEvent::from(&parser_task);
265        assert_eq!(parser_evt.account, "acc");
266        assert_eq!(parser_evt.platform, "pf");
267        assert_eq!(parser_evt.modules, Some(vec!["m1".to_string()]));
268
269        let error_task = TaskErrorEvent {
270            id: Uuid::now_v7(),
271            account_task: task_model,
272            error_msg: "err".to_string(),
273            timestamp: 0,
274            metadata: json!({"e":"x"}).as_object().cloned().unwrap_or_default(),
275            context: ExecutionMark::default(),
276            run_id,
277            prefix_request: Uuid::now_v7(),
278        };
279        let error_evt = ParserTaskModelEvent::from(&error_task);
280        assert_eq!(error_evt.account, "acc");
281        assert_eq!(error_evt.platform, "pf");
282        assert_eq!(error_evt.modules, Some(vec!["m1".to_string()]));
283    }
284
285    #[test]
286    fn parser_task_model_event_envelope_key_is_stable() {
287        let payload = json!({"account":"acc","platform":"pf"});
288        let evt = EventEnvelope::engine(EventType::ParserTaskModel, EventPhase::Completed, payload);
289        assert_eq!(evt.event_key(), "engine.parser_task_model.completed");
290    }
291}
292
293#[derive(Debug, Clone, Serialize, Deserialize)]
294pub struct TaskModelEvent {
295    pub account: String,
296    pub platform: String,
297    pub modules: Option<Vec<String>>,
298}
299impl From<&TaskEvent> for TaskModelEvent {
300    fn from(value: &TaskEvent) -> Self {
301        TaskModelEvent {
302            account: value.account.clone(),
303            platform: value.platform.clone(),
304            modules: None,
305        }
306    }
307}
308
309/// Compact event payload representing parser/error task-model inputs.
310#[derive(Debug, Clone, Serialize, Deserialize)]
311pub struct ParserTaskModelEvent {
312    pub account: String,
313    pub platform: String,
314    pub modules: Option<Vec<String>>,
315    #[serde(skip)]
316    pub metadata: Option<serde_json::Value>,
317}
318impl From<&TaskParserEvent> for ParserTaskModelEvent {
319    fn from(value: &TaskParserEvent) -> Self {
320        ParserTaskModelEvent {
321            account: value.account_task.account.clone(),
322            platform: value.account_task.platform.clone(),
323            modules: value.account_task.module.clone(),
324            metadata: Some(serde_json::to_value(value.metadata.clone()).unwrap_or_default()),
325        }
326    }
327}
328impl From<&TaskErrorEvent> for ParserTaskModelEvent {
329    fn from(value: &TaskErrorEvent) -> Self {
330        Self {
331            account: value.account_task.account.clone(),
332            platform: value.account_task.platform.clone(),
333            modules: value.account_task.module.clone(),
334            metadata: Some(serde_json::to_value(value.metadata.clone()).unwrap_or_default()),
335        }
336    }
337}
338
339/// Event payload emitted when a concrete module instance is generated.
340#[derive(Debug, Clone, Serialize, Deserialize)]
341pub struct ModuleGenerateEvent {
342    pub account: String,
343    pub platform: String,
344    pub module: String,
345}
346
347impl From<&Module> for ModuleGenerateEvent {
348    fn from(value: &Module) -> Self {
349        Self {
350            account: value.account.name.clone(),
351            platform: value.platform.name.clone(),
352            module: value.module.name().clone(),
353        }
354    }
355}
356
357/// Event payload for outbound request publication.
358#[derive(Debug, Clone, Serialize, Deserialize)]
359pub struct RequestEvent {
360    pub request_id: Uuid,
361    pub url: String,
362    pub account: String,
363    pub platform: String,
364    pub module: String,
365    pub method: String,
366    #[serde(skip)]
367    pub metadata: serde_json::Value,
368}
369
370impl From<&Request> for RequestEvent {
371    fn from(value: &Request) -> Self {
372        RequestEvent {
373            request_id: value.id,
374            url: value.url.clone(),
375            account: value.account.clone(),
376            platform: value.platform.clone(),
377            module: value.module.clone(),
378            method: value.method.clone(),
379            metadata: serde_json::to_value(value.meta.clone()).unwrap_or_default(),
380        }
381    }
382}
383
384/// Event payload for download execution telemetry.
385#[derive(Debug, Clone, Serialize, Deserialize)]
386pub struct DownloadEvent {
387    pub request_id: Uuid,
388    pub url: String,
389    pub account: String,
390    pub platform: String,
391    pub module: String,
392    pub method: String,
393    pub duration_ms: Option<u64>,
394    pub response_size: Option<u32>,
395    pub status_code: Option<u16>,
396}
397impl From<&Request> for DownloadEvent {
398    fn from(value: &Request) -> Self {
399        DownloadEvent {
400            request_id: value.id,
401            url: value.url.clone(),
402            account: value.account.clone(),
403            platform: value.platform.clone(),
404            module: value.module.clone(),
405            method: value.method.clone(),
406            duration_ms: None,
407            response_size: None,
408            status_code: None,
409        }
410    }
411}
412
413/// Event payload for parser execution telemetry.
414#[derive(Debug, Clone, Serialize, Deserialize)]
415pub struct ParserEvent {
416    pub account: String,
417    pub platform: String,
418    pub module: String,
419    pub request_id: String,
420}
421
422impl From<&Response> for ParserEvent {
423    fn from(value: &Response) -> Self {
424        Self {
425            account: value.account.clone(),
426            platform: value.platform.clone(),
427            module: value.module.clone(),
428            request_id: value.id.to_string(),
429        }
430    }
431}
432
433/// Event payload for data-middleware transformation stages.
434#[derive(Debug, Clone, Serialize, Deserialize)]
435pub struct DataMiddlewareEvent {
436    pub account: String,
437    pub platform: String,
438    pub module: String,
439    pub request_id: String,
440    pub schema_size: usize,
441    pub after_size: Option<usize>,
442}
443impl From<&crate::common::model::data::DataEvent> for DataMiddlewareEvent {
444    fn from(value: &crate::common::model::data::DataEvent) -> Self {
445        Self {
446            account: value.account.clone(),
447            platform: value.platform.clone(),
448            module: value.module.clone(),
449            request_id: value.request_id.to_string(),
450            schema_size: 0,
451            after_size: None,
452        }
453    }
454}
455
456/// Event payload for final data-store persistence stages.
457#[derive(Debug, Clone, Serialize, Deserialize)]
458pub struct DataStoreEvent {
459    pub account: String,
460    pub platform: String,
461    pub module: String,
462    pub request_id: String,
463    pub schema_size: (usize, usize),
464    pub store_middleware: Option<String>,
465}
466impl From<&crate::common::model::data::DataEvent> for DataStoreEvent {
467    fn from(value: &crate::common::model::data::DataEvent) -> Self {
468        Self {
469            account: value.account.clone(),
470            platform: value.platform.clone(),
471            module: value.module.clone(),
472            request_id: value.request_id.to_string(),
473            schema_size: (0, 0),
474            store_middleware: None,
475        }
476    }
477}
478
479/// Event payload for request middleware application.
480#[derive(Debug, Clone, Serialize, Deserialize)]
481pub struct RequestMiddlewareEvent {
482    pub request_id: Uuid,
483    pub url: String,
484    pub account: String,
485    pub platform: String,
486    pub module: String,
487    pub method: String,
488    pub middleware: Vec<String>,
489}
490impl From<&Request> for RequestMiddlewareEvent {
491    fn from(value: &Request) -> Self {
492        RequestMiddlewareEvent {
493            request_id: value.id,
494            url: value.url.clone(),
495            account: value.account.clone(),
496            platform: value.platform.clone(),
497            module: value.module.clone(),
498            method: value.method.clone(),
499            middleware: value.download_middleware.clone(),
500        }
501    }
502}
503
504/// Event payload for response middleware processing.
505#[derive(Debug, Clone, Serialize, Deserialize)]
506pub struct ResponseEvent {
507    pub response_id: Uuid,
508    pub account: String,
509    pub platform: String,
510    pub module: String,
511    pub status_code: Option<u16>,
512    pub middleware: Vec<String>,
513}
514impl From<&Response> for ResponseEvent {
515    fn from(value: &Response) -> Self {
516        ResponseEvent {
517            response_id: value.id,
518            account: value.account.clone(),
519            platform: value.platform.clone(),
520            module: value.module.clone(),
521            status_code: Some(value.status_code),
522            middleware: value.data_middleware.clone(),
523        }
524    }
525}
526
527/// Event payload for periodic component health reporting.
528#[derive(Debug, Clone, Serialize, Deserialize)]
529pub struct HealthCheckEvent {
530    pub component: String,
531    pub status: String,
532    pub metrics: serde_json::Value,
533    pub timestamp: u64,
534}
535impl From<&Response> for ModuleGenerateEvent {
536    fn from(value: &Response) -> Self {
537        Self {
538            account: value.account.clone(),
539            platform: value.platform.clone(),
540            module: value.module.clone(),
541        }
542    }
543}
mocra_core/engine/events/events.rs

mocra_core/engine/events/
events.rs
