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 DevicePaired {
136 id: String,
137 name: String,
138 },
139
140 DeviceRevoked {
141 id: String,
142 },
143
144 AlertRecorded {
145 id: String,
146 seq: i64,
147 },
148
149 RunLogProgress {
150 pipeline_id: String,
151 run_number: i64,
152 node_id: String,
153 received_bytes: u64,
154 },
155
156 RunLogReady {
157 pipeline_id: String,
158 run_number: i64,
159 node_id: String,
160 revision: i64,
161 },
162
163 RunLogFailed {
164 pipeline_id: String,
165 run_number: i64,
166 node_id: String,
167 message: String,
168 },
169
170 RunLogTreeChanged {
171 pipeline_id: String,
172 run_number: i64,
173 },
174
175 RunUpdated {
176 pipeline_id: String,
177 run_number: i64,
178 run: PipelineRun,
179 },
180}
181
182impl CoreEvent {
183 pub fn event_name(&self) -> &'static str {
184 match self {
185 CoreEvent::ProvidersChanged => "providers-changed",
186 CoreEvent::PipelineViewsChanged => "pipeline-views-changed",
187 CoreEvent::PipelinesUpdated { .. } => "pipelines-updated",
188 CoreEvent::PipelineStatusChanged { .. } => "pipeline-status-changed",
189 CoreEvent::PipelinesFailed { .. } => "pipelines-failed",
190 CoreEvent::RunTriggered { .. } => "run-triggered",
191 CoreEvent::RunCancelled { .. } => "run-cancelled",
192 CoreEvent::MetricsGenerated { .. } => "metrics-generated",
193 CoreEvent::MetricsGlobalConfigChanged => "metrics-global-config-changed",
194 CoreEvent::MetricsConfigChanged { .. } => "metrics-config-changed",
195 CoreEvent::MetricsFlushed { .. } => "metrics-flushed",
196 CoreEvent::ProviderStatusUpdated { .. } => "provider-status-updated",
197 CoreEvent::ProviderHealthAlert { .. } => "provider-health-alert",
198 CoreEvent::MigrationProgress { .. } => "migration-progress",
199 CoreEvent::MigrationComplete { .. } => "migration-complete",
200 CoreEvent::ProviderAdded { .. } => "provider-added",
201 CoreEvent::ProviderUpdated { .. } => "provider-updated",
202 CoreEvent::ProviderRemoved { .. } => "provider-removed",
203 CoreEvent::PipelineCacheInvalidated { .. } => "pipeline-cache-invalidated",
204 CoreEvent::RunHistoryCacheInvalidated { .. } => "run-history-cache-invalidated",
205 CoreEvent::VaultUnlocked => "vault-unlocked",
206 CoreEvent::VaultLocked => "vault-locked",
207 CoreEvent::DevicePaired { .. } => "device-paired",
208 CoreEvent::DeviceRevoked { .. } => "device-revoked",
209 CoreEvent::AlertRecorded { .. } => "alert-recorded",
210 CoreEvent::RunLogProgress { .. } => "run-log-progress",
211 CoreEvent::RunLogReady { .. } => "run-log-ready",
212 CoreEvent::RunLogFailed { .. } => "run-log-failed",
213 CoreEvent::RunLogTreeChanged { .. } => "run-log-tree-changed",
214 CoreEvent::RunUpdated { .. } => "run-updated",
215 }
216 }
217
218 pub fn to_json_payload(&self) -> serde_json::Value {
219 match self {
220 CoreEvent::ProvidersChanged => serde_json::json!({}),
221 CoreEvent::PipelineViewsChanged => serde_json::json!({}),
222 CoreEvent::PipelinesUpdated {
223 pipelines,
224 provider_id,
225 timestamp,
226 } => serde_json::json!({
227 "pipelines": pipelines,
228 "providerId": provider_id,
229 "timestamp": timestamp,
230 }),
231 CoreEvent::PipelineStatusChanged { pipeline_ids } => serde_json::json!({
232 "pipeline_ids": pipeline_ids,
233 }),
234 CoreEvent::PipelinesFailed { pipelines } => serde_json::json!({
235 "pipelines": pipelines,
236 }),
237 CoreEvent::RunTriggered { workflow_id } => serde_json::json!(workflow_id),
238 CoreEvent::RunCancelled { pipeline_id } => serde_json::json!(pipeline_id),
239 CoreEvent::MetricsGenerated { pipeline_id } => serde_json::json!(pipeline_id),
240 CoreEvent::MetricsGlobalConfigChanged => serde_json::json!({}),
241 CoreEvent::MetricsConfigChanged { pipeline_id } => serde_json::json!(pipeline_id),
242 CoreEvent::MetricsFlushed { pipeline_id } => {
243 let mut json = serde_json::json!({});
244 if let Some(pid) = pipeline_id {
245 json["pipeline_id"] = serde_json::json!(pid);
246 }
247 json
248 }
249 CoreEvent::ProviderStatusUpdated { provider } => serde_json::json!({
250 "provider": provider,
251 }),
252 CoreEvent::ProviderHealthAlert {
253 alert_id,
254 notification_key,
255 provider_ids,
256 level,
257 title,
258 body,
259 link,
260 recovered,
261 ..
262 } => serde_json::json!({
263 "alert_id": alert_id,
264 "notification_key": notification_key,
265 "provider_ids": provider_ids,
266 "level": level,
267 "title": title,
268 "body": body,
269 "link": link,
270 "recovered": recovered,
271 }),
272 CoreEvent::MigrationProgress {
273 step,
274 step_index,
275 total_steps,
276 message,
277 } => serde_json::json!({
278 "step": step,
279 "step_index": step_index,
280 "total_steps": total_steps,
281 "message": message,
282 }),
283 CoreEvent::MigrationComplete { success, summary } => serde_json::json!({
284 "success": success,
285 "summary": summary,
286 }),
287 CoreEvent::ProviderAdded {
288 provider,
289 timestamp,
290 } => serde_json::json!({
291 "provider": provider,
292 "action": "added",
293 "timestamp": timestamp,
294 }),
295 CoreEvent::ProviderUpdated {
296 provider,
297 timestamp,
298 } => serde_json::json!({
299 "provider": provider,
300 "action": "updated",
301 "timestamp": timestamp,
302 }),
303 CoreEvent::ProviderRemoved {
304 provider,
305 timestamp,
306 } => serde_json::json!({
307 "provider": provider,
308 "action": "removed",
309 "timestamp": timestamp,
310 }),
311 CoreEvent::PipelineCacheInvalidated {
312 provider_id,
313 reason,
314 } => {
315 let mut json = serde_json::json!({
316 "reason": reason,
317 });
318 if let Some(pid) = provider_id {
319 json["providerId"] = serde_json::json!(pid);
320 }
321 json
322 }
323 CoreEvent::RunHistoryCacheInvalidated { pipeline_id } => {
324 let mut json = serde_json::json!({});
325 if let Some(pid) = pipeline_id {
326 json["pipeline_id"] = serde_json::json!(pid);
327 }
328 json
329 }
330 CoreEvent::VaultUnlocked => serde_json::json!({}),
331 CoreEvent::VaultLocked => serde_json::json!({}),
332 CoreEvent::DevicePaired { id, name } => serde_json::json!({
333 "id": id,
334 "name": name,
335 }),
336 CoreEvent::DeviceRevoked { id } => serde_json::json!({
337 "id": id,
338 }),
339 CoreEvent::AlertRecorded { id, seq } => serde_json::json!({
340 "id": id,
341 "seq": seq,
342 }),
343 CoreEvent::RunLogProgress {
344 pipeline_id,
345 run_number,
346 node_id,
347 received_bytes,
348 } => serde_json::json!({
349 "pipelineId": pipeline_id,
350 "runNumber": run_number,
351 "nodeId": node_id,
352 "receivedBytes": received_bytes,
353 }),
354 CoreEvent::RunLogReady {
355 pipeline_id,
356 run_number,
357 node_id,
358 revision,
359 } => serde_json::json!({
360 "pipelineId": pipeline_id,
361 "runNumber": run_number,
362 "nodeId": node_id,
363 "revision": revision,
364 }),
365 CoreEvent::RunLogFailed {
366 pipeline_id,
367 run_number,
368 node_id,
369 message,
370 } => serde_json::json!({
371 "pipelineId": pipeline_id,
372 "runNumber": run_number,
373 "nodeId": node_id,
374 "message": message,
375 }),
376 CoreEvent::RunLogTreeChanged {
377 pipeline_id,
378 run_number,
379 } => serde_json::json!({
380 "pipelineId": pipeline_id,
381 "runNumber": run_number,
382 }),
383 CoreEvent::RunUpdated {
384 pipeline_id,
385 run_number,
386 run,
387 } => serde_json::json!({
388 "pipelineId": pipeline_id,
389 "runNumber": run_number,
390 "run": run,
391 }),
392 }
393 }
394}
395
396#[async_trait]
397pub trait EventBus: Send + Sync {
398 async fn emit(&self, event: CoreEvent);
399
400 async fn emit_to(&self, target: &str, event: CoreEvent);
401}
402
403pub struct NoOpEventBus;
404
405#[async_trait]
406impl EventBus for NoOpEventBus {
407 async fn emit(&self, _event: CoreEvent) {}
408
409 async fn emit_to(&self, _target: &str, _event: CoreEvent) {}
410}
411
412#[cfg(test)]
413mod tests {
414 use super::*;
415
416 #[test]
417 fn test_event_names() {
418 assert_eq!(
419 CoreEvent::ProvidersChanged.event_name(),
420 "providers-changed"
421 );
422 assert_eq!(
423 CoreEvent::PipelineViewsChanged.event_name(),
424 "pipeline-views-changed"
425 );
426 assert_eq!(
427 CoreEvent::PipelineStatusChanged {
428 pipeline_ids: vec!["a".to_string()]
429 }
430 .event_name(),
431 "pipeline-status-changed"
432 );
433 }
434
435 #[test]
436 fn test_event_serialization() {
437 let event = CoreEvent::PipelineStatusChanged {
438 pipeline_ids: vec!["github__1__web".to_string(), "github__1__api".to_string()],
439 };
440 let json = serde_json::to_string(&event).unwrap();
441 assert!(json.contains("PipelineStatusChanged"));
442 assert!(json.contains("github__1__web"));
443 assert_eq!(
444 event.to_json_payload(),
445 serde_json::json!({ "pipeline_ids": ["github__1__web", "github__1__api"] })
446 );
447 }
448
449 #[test]
450 fn test_provider_health_alert_wire_format() {
451 let alert = |link| CoreEvent::ProviderHealthAlert {
452 alert_id: "alert-7".to_string(),
453 alert_key: "incident:7".to_string(),
454 notification_key: "status-page:feed".to_string(),
455 provider_ids: vec![1, 2],
456 level: HealthLevel::MajorOutage,
457 title: "GitHub is down".to_string(),
458 body: "Investigating".to_string(),
459 link,
460 recovered: false,
461 };
462 let with_link = alert(Some(AlertLink {
463 url: "https://www.githubstatus.com/incidents/abc".to_string(),
464 kind: AlertLinkKind::Incident,
465 }));
466
467 assert_eq!(with_link.event_name(), "provider-health-alert");
468 assert_eq!(
469 with_link.to_json_payload(),
470 serde_json::json!({
471 "alert_id": "alert-7",
472 "notification_key": "status-page:feed",
473 "provider_ids": [1, 2],
474 "level": "major_outage",
475 "title": "GitHub is down",
476 "body": "Investigating",
477 "link": {
478 "url": "https://www.githubstatus.com/incidents/abc",
479 "kind": "incident"
480 },
481 "recovered": false,
482 })
483 );
484 assert_eq!(
485 alert(None).to_json_payload()["link"],
486 serde_json::Value::Null
487 );
488 }
489
490 #[test]
491 fn test_vault_lock_and_pipelines_failed_wire_format() {
492 assert_eq!(CoreEvent::VaultLocked.event_name(), "vault-locked");
493 assert_eq!(
494 CoreEvent::VaultLocked.to_json_payload(),
495 serde_json::json!({})
496 );
497
498 let failed = CoreEvent::PipelinesFailed {
499 pipelines: Vec::new(),
500 };
501 assert_eq!(failed.event_name(), "pipelines-failed");
502 assert_eq!(
503 failed.to_json_payload(),
504 serde_json::json!({ "pipelines": [] })
505 );
506
507 let paired = CoreEvent::DevicePaired {
508 id: "d1".into(),
509 name: "Pixel".into(),
510 };
511 assert_eq!(paired.event_name(), "device-paired");
512 assert_eq!(
513 paired.to_json_payload(),
514 serde_json::json!({ "id": "d1", "name": "Pixel" })
515 );
516
517 let revoked = CoreEvent::DeviceRevoked { id: "d1".into() };
518 assert_eq!(revoked.event_name(), "device-revoked");
519 assert_eq!(revoked.to_json_payload(), serde_json::json!({ "id": "d1" }));
520
521 let recorded = CoreEvent::AlertRecorded {
522 id: "a1".into(),
523 seq: 7,
524 };
525 assert_eq!(recorded.event_name(), "alert-recorded");
526 assert_eq!(
527 recorded.to_json_payload(),
528 serde_json::json!({ "id": "a1", "seq": 7 })
529 );
530 }
531}