pipedash_core/
event.rs

1use async_trait::async_trait;
2use serde::{
3    Deserialize,
4    Serialize,
5};
6
7use crate::domain::{
8    provider::ProviderSummary,
9    Pipeline,
10};
11
12#[derive(Debug, Clone, Serialize, Deserialize)]
13#[serde(rename_all = "kebab-case")]
14pub enum CacheInvalidationReason {
15    Fetch,
16    ProviderChange,
17    ManualRefresh,
18}
19
20#[derive(Debug, Clone, Serialize, Deserialize)]
21#[serde(tag = "type", content = "payload")]
22pub enum CoreEvent {
23    ProvidersChanged,
24
25    PipelinesFetched {
26        provider_id: i64,
27    },
28
29    PipelinesFetchError {
30        error: String,
31    },
32
33    PipelinesUpdated {
34        pipelines: Vec<Pipeline>,
35        provider_id: Option<i64>,
36        timestamp: i64,
37    },
38
39    PipelineStatusChanged {
40        pipelines: Vec<Pipeline>,
41    },
42
43    RunTriggered {
44        workflow_id: String,
45    },
46
47    RunCancelled {
48        pipeline_id: String,
49    },
50
51    RefreshError {
52        error: String,
53    },
54
55    MetricsGenerated {
56        pipeline_id: String,
57    },
58
59    MetricsGlobalConfigChanged,
60
61    MetricsConfigChanged {
62        pipeline_id: String,
63    },
64
65    MetricsFlushed {
66        pipeline_id: Option<String>,
67    },
68
69    ProviderStatusUpdated {
70        provider_id: i64,
71    },
72
73    MigrationProgress {
74        step: String,
75        step_index: usize,
76        total_steps: usize,
77        message: String,
78    },
79
80    MigrationComplete {
81        success: bool,
82        summary: String,
83    },
84
85    ProviderAdded {
86        provider: ProviderSummary,
87        timestamp: i64,
88    },
89
90    ProviderUpdated {
91        provider: ProviderSummary,
92        timestamp: i64,
93    },
94
95    ProviderRemoved {
96        provider: ProviderSummary,
97        timestamp: i64,
98    },
99
100    PipelineCacheInvalidated {
101        provider_id: Option<i64>,
102        reason: CacheInvalidationReason,
103    },
104
105    RunHistoryCacheInvalidated {
106        pipeline_id: Option<String>,
107    },
108
109    VaultUnlocked,
110}
111
112impl CoreEvent {
113    pub fn event_name(&self) -> &'static str {
114        match self {
115            CoreEvent::ProvidersChanged => "providers-changed",
116            CoreEvent::PipelinesFetched { .. } => "pipelines-fetched",
117            CoreEvent::PipelinesFetchError { .. } => "pipelines-fetch-error",
118            CoreEvent::PipelinesUpdated { .. } => "pipelines-updated",
119            CoreEvent::PipelineStatusChanged { .. } => "pipeline-status-changed",
120            CoreEvent::RunTriggered { .. } => "run-triggered",
121            CoreEvent::RunCancelled { .. } => "run-cancelled",
122            CoreEvent::RefreshError { .. } => "refresh-error",
123            CoreEvent::MetricsGenerated { .. } => "metrics-generated",
124            CoreEvent::MetricsGlobalConfigChanged => "metrics-global-config-changed",
125            CoreEvent::MetricsConfigChanged { .. } => "metrics-config-changed",
126            CoreEvent::MetricsFlushed { .. } => "metrics-flushed",
127            CoreEvent::ProviderStatusUpdated { .. } => "provider-status-updated",
128            CoreEvent::MigrationProgress { .. } => "migration-progress",
129            CoreEvent::MigrationComplete { .. } => "migration-complete",
130            CoreEvent::ProviderAdded { .. } => "provider-added",
131            CoreEvent::ProviderUpdated { .. } => "provider-updated",
132            CoreEvent::ProviderRemoved { .. } => "provider-removed",
133            CoreEvent::PipelineCacheInvalidated { .. } => "pipeline-cache-invalidated",
134            CoreEvent::RunHistoryCacheInvalidated { .. } => "run-history-cache-invalidated",
135            CoreEvent::VaultUnlocked => "vault-unlocked",
136        }
137    }
138
139    pub fn to_json_payload(&self) -> serde_json::Value {
140        match self {
141            CoreEvent::ProvidersChanged => serde_json::json!({}),
142            CoreEvent::PipelinesFetched { provider_id } => serde_json::json!(provider_id),
143            CoreEvent::PipelinesFetchError { error } => serde_json::json!(error),
144            CoreEvent::PipelinesUpdated {
145                pipelines,
146                provider_id,
147                timestamp,
148            } => {
149                let mut json = serde_json::json!({
150                    "pipelines": pipelines,
151                    "timestamp": timestamp,
152                });
153                if let Some(pid) = provider_id {
154                    json["providerId"] = serde_json::json!(pid);
155                }
156                json
157            }
158            CoreEvent::PipelineStatusChanged { pipelines } => {
159                serde_json::to_value(pipelines).unwrap_or_default()
160            }
161            CoreEvent::RunTriggered { workflow_id } => serde_json::json!(workflow_id),
162            CoreEvent::RunCancelled { pipeline_id } => serde_json::json!(pipeline_id),
163            CoreEvent::RefreshError { error } => serde_json::json!(error),
164            CoreEvent::MetricsGenerated { pipeline_id } => serde_json::json!(pipeline_id),
165            CoreEvent::MetricsGlobalConfigChanged => serde_json::json!({}),
166            CoreEvent::MetricsConfigChanged { pipeline_id } => serde_json::json!(pipeline_id),
167            CoreEvent::MetricsFlushed { pipeline_id } => {
168                let mut json = serde_json::json!({});
169                if let Some(pid) = pipeline_id {
170                    json["pipeline_id"] = serde_json::json!(pid);
171                }
172                json
173            }
174            CoreEvent::ProviderStatusUpdated { provider_id } => serde_json::json!(provider_id),
175            CoreEvent::MigrationProgress {
176                step,
177                step_index,
178                total_steps,
179                message,
180            } => serde_json::json!({
181                "step": step,
182                "step_index": step_index,
183                "total_steps": total_steps,
184                "message": message,
185            }),
186            CoreEvent::MigrationComplete { success, summary } => serde_json::json!({
187                "success": success,
188                "summary": summary,
189            }),
190            CoreEvent::ProviderAdded {
191                provider,
192                timestamp,
193            } => serde_json::json!({
194                "provider": provider,
195                "action": "added",
196                "timestamp": timestamp,
197            }),
198            CoreEvent::ProviderUpdated {
199                provider,
200                timestamp,
201            } => serde_json::json!({
202                "provider": provider,
203                "action": "updated",
204                "timestamp": timestamp,
205            }),
206            CoreEvent::ProviderRemoved {
207                provider,
208                timestamp,
209            } => serde_json::json!({
210                "provider": provider,
211                "action": "removed",
212                "timestamp": timestamp,
213            }),
214            CoreEvent::PipelineCacheInvalidated {
215                provider_id,
216                reason,
217            } => {
218                let mut json = serde_json::json!({
219                    "reason": reason,
220                });
221                if let Some(pid) = provider_id {
222                    json["providerId"] = serde_json::json!(pid);
223                }
224                json
225            }
226            CoreEvent::RunHistoryCacheInvalidated { pipeline_id } => {
227                let mut json = serde_json::json!({});
228                if let Some(pid) = pipeline_id {
229                    json["pipeline_id"] = serde_json::json!(pid);
230                }
231                json
232            }
233            CoreEvent::VaultUnlocked => serde_json::json!({}),
234        }
235    }
236}
237
238#[async_trait]
239pub trait EventBus: Send + Sync {
240    async fn emit(&self, event: CoreEvent);
241
242    async fn emit_to(&self, target: &str, event: CoreEvent);
243}
244
245pub struct NoOpEventBus;
246
247#[async_trait]
248impl EventBus for NoOpEventBus {
249    async fn emit(&self, _event: CoreEvent) {}
250
251    async fn emit_to(&self, _target: &str, _event: CoreEvent) {}
252}
253
254#[cfg(test)]
255mod tests {
256    use super::*;
257
258    #[test]
259    fn test_event_names() {
260        assert_eq!(
261            CoreEvent::ProvidersChanged.event_name(),
262            "providers-changed"
263        );
264        assert_eq!(
265            CoreEvent::PipelinesFetched { provider_id: 1 }.event_name(),
266            "pipelines-fetched"
267        );
268    }
269
270    #[test]
271    fn test_event_serialization() {
272        let event = CoreEvent::PipelinesFetched { provider_id: 42 };
273        let json = serde_json::to_string(&event).unwrap();
274        assert!(json.contains("PipelinesFetched"));
275        assert!(json.contains("42"));
276    }
277}
pipedash_core/event.rs

pipedash_core/
event.rs
