1use async_trait::async_trait;
2use serde::{
3 Deserialize,
4 Serialize,
5};
6
7use crate::domain::{
8 provider::ProviderSummary,
9 HealthLevel,
10 Pipeline,
11 PipelineRun,
12};
13
14#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
15#[serde(rename_all = "snake_case")]
16pub enum AlertLinkKind {
17 Incident,
18 StatusPage,
19}
20
21#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
22pub struct AlertLink {
23 pub url: String,
24 pub kind: AlertLinkKind,
25}
26
27#[derive(Debug, Clone, Serialize, Deserialize)]
28#[serde(rename_all = "kebab-case")]
29pub enum CacheInvalidationReason {
30 Fetch,
31 ProviderChange,
32 ManualRefresh,
33}
34
35#[derive(Debug, Clone, Serialize, Deserialize)]
36#[serde(tag = "type", content = "payload")]
37pub enum CoreEvent {
38 ProvidersChanged,
39
40 PipelineViewsChanged,
41
42 PipelinesUpdated {
43 pipelines: Vec<Pipeline>,
44 provider_id: i64,
45 timestamp: i64,
46 },
47
48 PipelineStatusChanged {
49 pipeline_ids: Vec<String>,
50 },
51
52 PipelinesFailed {
53 pipelines: Vec<Pipeline>,
54 },
55
56 RunTriggered {
57 workflow_id: String,
58 },
59
60 RunCancelled {
61 pipeline_id: String,
62 },
63
64 MetricsGenerated {
65 pipeline_id: String,
66 },
67
68 MetricsGlobalConfigChanged,
69
70 MetricsConfigChanged {
71 pipeline_id: String,
72 },
73
74 MetricsFlushed {
75 pipeline_id: Option<String>,
76 },
77
78 ProviderStatusUpdated {
79 provider: ProviderSummary,
80 },
81
82 ProviderHealthAlert {
83 alert_id: String,
84 #[serde(default)]
85 alert_key: String,
86 notification_key: String,
87 provider_ids: Vec<i64>,
88 level: HealthLevel,
89 title: String,
90 body: String,
91 link: Option<AlertLink>,
92 recovered: bool,
93 },
94
95 MigrationProgress {
96 step: String,
97 step_index: usize,
98 total_steps: usize,
99 message: String,
100 },
101
102 MigrationComplete {
103 success: bool,
104 summary: String,
105 },
106
107 ProviderAdded {
108 provider: ProviderSummary,
109 timestamp: i64,
110 },
111
112 ProviderUpdated {
113 provider: ProviderSummary,
114 timestamp: i64,
115 },
116
117 ProviderRemoved {
118 provider: ProviderSummary,
119 timestamp: i64,
120 },
121
122 PipelineCacheInvalidated {
123 provider_id: Option<i64>,
124 reason: CacheInvalidationReason,
125 },
126
127 RunHistoryCacheInvalidated {
128 pipeline_id: Option<String>,
129 },
130
131 VaultUnlocked,
132
133 VaultLocked,
134
135 RunLogProgress {
136 pipeline_id: String,
137 run_number: i64,
138 node_id: String,
139 received_bytes: u64,
140 },
141
142 RunLogReady {
143 pipeline_id: String,
144 run_number: i64,
145 node_id: String,
146 revision: i64,
147 },
148
149 RunLogFailed {
150 pipeline_id: String,
151 run_number: i64,
152 node_id: String,
153 message: String,
154 },
155
156 RunLogTreeChanged {
157 pipeline_id: String,
158 run_number: i64,
159 },
160
161 RunUpdated {
162 pipeline_id: String,
163 run_number: i64,
164 run: PipelineRun,
165 },
166}
167
168impl CoreEvent {
169 pub fn event_name(&self) -> &'static str {
170 match self {
171 CoreEvent::ProvidersChanged => "providers-changed",
172 CoreEvent::PipelineViewsChanged => "pipeline-views-changed",
173 CoreEvent::PipelinesUpdated { .. } => "pipelines-updated",
174 CoreEvent::PipelineStatusChanged { .. } => "pipeline-status-changed",
175 CoreEvent::PipelinesFailed { .. } => "pipelines-failed",
176 CoreEvent::RunTriggered { .. } => "run-triggered",
177 CoreEvent::RunCancelled { .. } => "run-cancelled",
178 CoreEvent::MetricsGenerated { .. } => "metrics-generated",
179 CoreEvent::MetricsGlobalConfigChanged => "metrics-global-config-changed",
180 CoreEvent::MetricsConfigChanged { .. } => "metrics-config-changed",
181 CoreEvent::MetricsFlushed { .. } => "metrics-flushed",
182 CoreEvent::ProviderStatusUpdated { .. } => "provider-status-updated",
183 CoreEvent::ProviderHealthAlert { .. } => "provider-health-alert",
184 CoreEvent::MigrationProgress { .. } => "migration-progress",
185 CoreEvent::MigrationComplete { .. } => "migration-complete",
186 CoreEvent::ProviderAdded { .. } => "provider-added",
187 CoreEvent::ProviderUpdated { .. } => "provider-updated",
188 CoreEvent::ProviderRemoved { .. } => "provider-removed",
189 CoreEvent::PipelineCacheInvalidated { .. } => "pipeline-cache-invalidated",
190 CoreEvent::RunHistoryCacheInvalidated { .. } => "run-history-cache-invalidated",
191 CoreEvent::VaultUnlocked => "vault-unlocked",
192 CoreEvent::VaultLocked => "vault-locked",
193 CoreEvent::RunLogProgress { .. } => "run-log-progress",
194 CoreEvent::RunLogReady { .. } => "run-log-ready",
195 CoreEvent::RunLogFailed { .. } => "run-log-failed",
196 CoreEvent::RunLogTreeChanged { .. } => "run-log-tree-changed",
197 CoreEvent::RunUpdated { .. } => "run-updated",
198 }
199 }
200
201 pub fn to_json_payload(&self) -> serde_json::Value {
202 match self {
203 CoreEvent::ProvidersChanged => serde_json::json!({}),
204 CoreEvent::PipelineViewsChanged => serde_json::json!({}),
205 CoreEvent::PipelinesUpdated {
206 pipelines,
207 provider_id,
208 timestamp,
209 } => serde_json::json!({
210 "pipelines": pipelines,
211 "providerId": provider_id,
212 "timestamp": timestamp,
213 }),
214 CoreEvent::PipelineStatusChanged { pipeline_ids } => serde_json::json!({
215 "pipeline_ids": pipeline_ids,
216 }),
217 CoreEvent::PipelinesFailed { pipelines } => serde_json::json!({
218 "pipelines": pipelines,
219 }),
220 CoreEvent::RunTriggered { workflow_id } => serde_json::json!(workflow_id),
221 CoreEvent::RunCancelled { pipeline_id } => serde_json::json!(pipeline_id),
222 CoreEvent::MetricsGenerated { pipeline_id } => serde_json::json!(pipeline_id),
223 CoreEvent::MetricsGlobalConfigChanged => serde_json::json!({}),
224 CoreEvent::MetricsConfigChanged { pipeline_id } => serde_json::json!(pipeline_id),
225 CoreEvent::MetricsFlushed { pipeline_id } => {
226 let mut json = serde_json::json!({});
227 if let Some(pid) = pipeline_id {
228 json["pipeline_id"] = serde_json::json!(pid);
229 }
230 json
231 }
232 CoreEvent::ProviderStatusUpdated { provider } => serde_json::json!({
233 "provider": provider,
234 }),
235 CoreEvent::ProviderHealthAlert {
236 alert_id,
237 notification_key,
238 provider_ids,
239 level,
240 title,
241 body,
242 link,
243 recovered,
244 ..
245 } => serde_json::json!({
246 "alert_id": alert_id,
247 "notification_key": notification_key,
248 "provider_ids": provider_ids,
249 "level": level,
250 "title": title,
251 "body": body,
252 "link": link,
253 "recovered": recovered,
254 }),
255 CoreEvent::MigrationProgress {
256 step,
257 step_index,
258 total_steps,
259 message,
260 } => serde_json::json!({
261 "step": step,
262 "step_index": step_index,
263 "total_steps": total_steps,
264 "message": message,
265 }),
266 CoreEvent::MigrationComplete { success, summary } => serde_json::json!({
267 "success": success,
268 "summary": summary,
269 }),
270 CoreEvent::ProviderAdded {
271 provider,
272 timestamp,
273 } => serde_json::json!({
274 "provider": provider,
275 "action": "added",
276 "timestamp": timestamp,
277 }),
278 CoreEvent::ProviderUpdated {
279 provider,
280 timestamp,
281 } => serde_json::json!({
282 "provider": provider,
283 "action": "updated",
284 "timestamp": timestamp,
285 }),
286 CoreEvent::ProviderRemoved {
287 provider,
288 timestamp,
289 } => serde_json::json!({
290 "provider": provider,
291 "action": "removed",
292 "timestamp": timestamp,
293 }),
294 CoreEvent::PipelineCacheInvalidated {
295 provider_id,
296 reason,
297 } => {
298 let mut json = serde_json::json!({
299 "reason": reason,
300 });
301 if let Some(pid) = provider_id {
302 json["providerId"] = serde_json::json!(pid);
303 }
304 json
305 }
306 CoreEvent::RunHistoryCacheInvalidated { pipeline_id } => {
307 let mut json = serde_json::json!({});
308 if let Some(pid) = pipeline_id {
309 json["pipeline_id"] = serde_json::json!(pid);
310 }
311 json
312 }
313 CoreEvent::VaultUnlocked => serde_json::json!({}),
314 CoreEvent::VaultLocked => serde_json::json!({}),
315 CoreEvent::RunLogProgress {
316 pipeline_id,
317 run_number,
318 node_id,
319 received_bytes,
320 } => serde_json::json!({
321 "pipelineId": pipeline_id,
322 "runNumber": run_number,
323 "nodeId": node_id,
324 "receivedBytes": received_bytes,
325 }),
326 CoreEvent::RunLogReady {
327 pipeline_id,
328 run_number,
329 node_id,
330 revision,
331 } => serde_json::json!({
332 "pipelineId": pipeline_id,
333 "runNumber": run_number,
334 "nodeId": node_id,
335 "revision": revision,
336 }),
337 CoreEvent::RunLogFailed {
338 pipeline_id,
339 run_number,
340 node_id,
341 message,
342 } => serde_json::json!({
343 "pipelineId": pipeline_id,
344 "runNumber": run_number,
345 "nodeId": node_id,
346 "message": message,
347 }),
348 CoreEvent::RunLogTreeChanged {
349 pipeline_id,
350 run_number,
351 } => serde_json::json!({
352 "pipelineId": pipeline_id,
353 "runNumber": run_number,
354 }),
355 CoreEvent::RunUpdated {
356 pipeline_id,
357 run_number,
358 run,
359 } => serde_json::json!({
360 "pipelineId": pipeline_id,
361 "runNumber": run_number,
362 "run": run,
363 }),
364 }
365 }
366}
367
368#[async_trait]
369pub trait EventBus: Send + Sync {
370 async fn emit(&self, event: CoreEvent);
371
372 async fn emit_to(&self, target: &str, event: CoreEvent);
373}
374
375pub struct NoOpEventBus;
376
377#[async_trait]
378impl EventBus for NoOpEventBus {
379 async fn emit(&self, _event: CoreEvent) {}
380
381 async fn emit_to(&self, _target: &str, _event: CoreEvent) {}
382}
383
384#[cfg(test)]
385mod tests {
386 use super::*;
387
388 #[test]
389 fn test_event_names() {
390 assert_eq!(
391 CoreEvent::ProvidersChanged.event_name(),
392 "providers-changed"
393 );
394 assert_eq!(
395 CoreEvent::PipelineViewsChanged.event_name(),
396 "pipeline-views-changed"
397 );
398 assert_eq!(
399 CoreEvent::PipelineStatusChanged {
400 pipeline_ids: vec!["a".to_string()]
401 }
402 .event_name(),
403 "pipeline-status-changed"
404 );
405 }
406
407 #[test]
408 fn test_event_serialization() {
409 let event = CoreEvent::PipelineStatusChanged {
410 pipeline_ids: vec!["github__1__web".to_string(), "github__1__api".to_string()],
411 };
412 let json = serde_json::to_string(&event).unwrap();
413 assert!(json.contains("PipelineStatusChanged"));
414 assert!(json.contains("github__1__web"));
415 assert_eq!(
416 event.to_json_payload(),
417 serde_json::json!({ "pipeline_ids": ["github__1__web", "github__1__api"] })
418 );
419 }
420
421 #[test]
422 fn test_provider_health_alert_wire_format() {
423 let alert = |link| CoreEvent::ProviderHealthAlert {
424 alert_id: "alert-7".to_string(),
425 alert_key: "incident:7".to_string(),
426 notification_key: "status-page:feed".to_string(),
427 provider_ids: vec![1, 2],
428 level: HealthLevel::MajorOutage,
429 title: "GitHub is down".to_string(),
430 body: "Investigating".to_string(),
431 link,
432 recovered: false,
433 };
434 let with_link = alert(Some(AlertLink {
435 url: "https://www.githubstatus.com/incidents/abc".to_string(),
436 kind: AlertLinkKind::Incident,
437 }));
438
439 assert_eq!(with_link.event_name(), "provider-health-alert");
440 assert_eq!(
441 with_link.to_json_payload(),
442 serde_json::json!({
443 "alert_id": "alert-7",
444 "notification_key": "status-page:feed",
445 "provider_ids": [1, 2],
446 "level": "major_outage",
447 "title": "GitHub is down",
448 "body": "Investigating",
449 "link": {
450 "url": "https://www.githubstatus.com/incidents/abc",
451 "kind": "incident"
452 },
453 "recovered": false,
454 })
455 );
456 assert_eq!(
457 alert(None).to_json_payload()["link"],
458 serde_json::Value::Null
459 );
460 }
461
462 #[test]
463 fn test_vault_lock_and_pipelines_failed_wire_format() {
464 assert_eq!(CoreEvent::VaultLocked.event_name(), "vault-locked");
465 assert_eq!(
466 CoreEvent::VaultLocked.to_json_payload(),
467 serde_json::json!({})
468 );
469
470 let failed = CoreEvent::PipelinesFailed {
471 pipelines: Vec::new(),
472 };
473 assert_eq!(failed.event_name(), "pipelines-failed");
474 assert_eq!(
475 failed.to_json_payload(),
476 serde_json::json!({ "pipelines": [] })
477 );
478 }
479}