bpm_engine_runtime/
history_handler.rs

1//! HistoryHandler: append every EngineEvent to HistoryRepo for Trace UI.
2
3use async_trait::async_trait;
4use bpm_engine_core::EngineEvent;
5use std::time::{SystemTime, UNIX_EPOCH};
6
7use super::handler::{EngineContext, EventHandler};
8
9fn occurred_at_now() -> String {
10    SystemTime::now()
11        .duration_since(UNIX_EPOCH)
12        .unwrap()
13        .as_secs()
14        .to_string()
15}
16
17fn event_type_instance_payload(
18    event: &EngineEvent,
19) -> Option<(&'static str, String, serde_json::Value)> {
20    match event {
21        EngineEvent::ProcessStarted(p) => Some((
22            "ProcessStarted",
23            p.instance_id.clone(),
24            serde_json::to_value(p).unwrap_or_default(),
25        )),
26        EngineEvent::TokenArrived(p) => Some((
27            "TokenArrived",
28            p.instance_id.clone(),
29            serde_json::to_value(p).unwrap_or_default(),
30        )),
31        EngineEvent::TokenCompleted(p) => Some((
32            "TokenCompleted",
33            p.instance_id.clone(),
34            serde_json::to_value(p).unwrap_or_default(),
35        )),
36        EngineEvent::UserTaskCreated(p) => Some((
37            "UserTaskCreated",
38            p.instance_id.clone(),
39            serde_json::to_value(p).unwrap_or_default(),
40        )),
41        EngineEvent::UserTaskCompleted(p) => Some((
42            "UserTaskCompleted",
43            p.instance_id.clone(),
44            serde_json::to_value(p).unwrap_or_default(),
45        )),
46        EngineEvent::TimerFired(p) => {
47            // TimerFired payload has only timer_id and token_id; store with empty instance_id.
48            Some((
49                "TimerFired",
50                String::new(),
51                serde_json::to_value(p).unwrap_or_default(),
52            ))
53        }
54        EngineEvent::TimerScheduled(p) => Some((
55            "TimerScheduled",
56            p.instance_id.clone(),
57            serde_json::to_value(p).unwrap_or_default(),
58        )),
59        EngineEvent::TokenFailed(p) => Some((
60            "TokenFailed",
61            p.instance_id.clone(),
62            serde_json::to_value(p).unwrap_or_default(),
63        )),
64        EngineEvent::SagaStarted(p) => Some((
65            "SagaStarted",
66            p.instance_id.clone(),
67            serde_json::to_value(p).unwrap_or_default(),
68        )),
69        EngineEvent::SagaCompleted(p) => Some((
70            "SagaCompleted",
71            p.instance_id.clone(),
72            serde_json::to_value(p).unwrap_or_default(),
73        )),
74        EngineEvent::ProcessCompleted(p) => Some((
75            "ProcessCompleted",
76            p.instance_id.clone(),
77            serde_json::to_value(p).unwrap_or_default(),
78        )),
79    }
80}
81
82pub struct HistoryHandler;
83
84#[async_trait]
85impl EventHandler for HistoryHandler {
86    async fn handle(&self, event: &EngineEvent, ctx: &mut EngineContext) -> Vec<EngineEvent> {
87        let Some((event_type, instance_id, payload)) = event_type_instance_payload(event) else {
88            return vec![];
89        };
90        if instance_id.is_empty() && event_type != "TimerFired" {
91            return vec![];
92        }
93        let Some(history_repo) = ctx.history_repo.as_ref() else {
94            return vec![];
95        };
96        let occurred_at = occurred_at_now();
97        let _ = history_repo
98            .append(&instance_id, event_type, &payload, &occurred_at)
99            .await;
100        vec![]
101    }
102}
bpm_engine_runtime/history_handler.rs

bpm_engine_runtime/
history_handler.rs
