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