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