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 PipelinesUpdated {
40 pipelines: Vec<Pipeline>,
41 provider_id: i64,
42 timestamp: i64,
43 },
44
45 PipelineStatusChanged {
46 pipeline_ids: Vec<String>,
47 },
48
49 RunTriggered {
50 workflow_id: String,
51 },
52
53 RunCancelled {
54 pipeline_id: String,
55 },
56
57 MetricsGenerated {
58 pipeline_id: String,
59 },
60
61 MetricsGlobalConfigChanged,
62
63 MetricsConfigChanged {
64 pipeline_id: String,
65 },
66
67 MetricsFlushed {
68 pipeline_id: Option<String>,
69 },
70
71 ProviderStatusUpdated {
72 provider: ProviderSummary,
73 },
74
75 ProviderHealthAlert {
76 alert_id: String,
77 notification_key: String,
78 provider_ids: Vec<i64>,
79 level: HealthLevel,
80 title: String,
81 body: String,
82 link: Option<AlertLink>,
83 recovered: bool,
84 },
85
86 MigrationProgress {
87 step: String,
88 step_index: usize,
89 total_steps: usize,
90 message: String,
91 },
92
93 MigrationComplete {
94 success: bool,
95 summary: String,
96 },
97
98 ProviderAdded {
99 provider: ProviderSummary,
100 timestamp: i64,
101 },
102
103 ProviderUpdated {
104 provider: ProviderSummary,
105 timestamp: i64,
106 },
107
108 ProviderRemoved {
109 provider: ProviderSummary,
110 timestamp: i64,
111 },
112
113 PipelineCacheInvalidated {
114 provider_id: Option<i64>,
115 reason: CacheInvalidationReason,
116 },
117
118 RunHistoryCacheInvalidated {
119 pipeline_id: Option<String>,
120 },
121
122 VaultUnlocked,
123}
124
125impl CoreEvent {
126 pub fn event_name(&self) -> &'static str {
127 match self {
128 CoreEvent::ProvidersChanged => "providers-changed",
129 CoreEvent::PipelinesUpdated { .. } => "pipelines-updated",
130 CoreEvent::PipelineStatusChanged { .. } => "pipeline-status-changed",
131 CoreEvent::RunTriggered { .. } => "run-triggered",
132 CoreEvent::RunCancelled { .. } => "run-cancelled",
133 CoreEvent::MetricsGenerated { .. } => "metrics-generated",
134 CoreEvent::MetricsGlobalConfigChanged => "metrics-global-config-changed",
135 CoreEvent::MetricsConfigChanged { .. } => "metrics-config-changed",
136 CoreEvent::MetricsFlushed { .. } => "metrics-flushed",
137 CoreEvent::ProviderStatusUpdated { .. } => "provider-status-updated",
138 CoreEvent::ProviderHealthAlert { .. } => "provider-health-alert",
139 CoreEvent::MigrationProgress { .. } => "migration-progress",
140 CoreEvent::MigrationComplete { .. } => "migration-complete",
141 CoreEvent::ProviderAdded { .. } => "provider-added",
142 CoreEvent::ProviderUpdated { .. } => "provider-updated",
143 CoreEvent::ProviderRemoved { .. } => "provider-removed",
144 CoreEvent::PipelineCacheInvalidated { .. } => "pipeline-cache-invalidated",
145 CoreEvent::RunHistoryCacheInvalidated { .. } => "run-history-cache-invalidated",
146 CoreEvent::VaultUnlocked => "vault-unlocked",
147 }
148 }
149
150 pub fn to_json_payload(&self) -> serde_json::Value {
151 match self {
152 CoreEvent::ProvidersChanged => serde_json::json!({}),
153 CoreEvent::PipelinesUpdated {
154 pipelines,
155 provider_id,
156 timestamp,
157 } => serde_json::json!({
158 "pipelines": pipelines,
159 "providerId": provider_id,
160 "timestamp": timestamp,
161 }),
162 CoreEvent::PipelineStatusChanged { pipeline_ids } => serde_json::json!({
163 "pipeline_ids": pipeline_ids,
164 }),
165 CoreEvent::RunTriggered { workflow_id } => serde_json::json!(workflow_id),
166 CoreEvent::RunCancelled { pipeline_id } => serde_json::json!(pipeline_id),
167 CoreEvent::MetricsGenerated { pipeline_id } => serde_json::json!(pipeline_id),
168 CoreEvent::MetricsGlobalConfigChanged => serde_json::json!({}),
169 CoreEvent::MetricsConfigChanged { pipeline_id } => serde_json::json!(pipeline_id),
170 CoreEvent::MetricsFlushed { pipeline_id } => {
171 let mut json = serde_json::json!({});
172 if let Some(pid) = pipeline_id {
173 json["pipeline_id"] = serde_json::json!(pid);
174 }
175 json
176 }
177 CoreEvent::ProviderStatusUpdated { provider } => serde_json::json!({
178 "provider": provider,
179 }),
180 CoreEvent::ProviderHealthAlert {
181 alert_id,
182 notification_key,
183 provider_ids,
184 level,
185 title,
186 body,
187 link,
188 recovered,
189 } => serde_json::json!({
190 "alert_id": alert_id,
191 "notification_key": notification_key,
192 "provider_ids": provider_ids,
193 "level": level,
194 "title": title,
195 "body": body,
196 "link": link,
197 "recovered": recovered,
198 }),
199 CoreEvent::MigrationProgress {
200 step,
201 step_index,
202 total_steps,
203 message,
204 } => serde_json::json!({
205 "step": step,
206 "step_index": step_index,
207 "total_steps": total_steps,
208 "message": message,
209 }),
210 CoreEvent::MigrationComplete { success, summary } => serde_json::json!({
211 "success": success,
212 "summary": summary,
213 }),
214 CoreEvent::ProviderAdded {
215 provider,
216 timestamp,
217 } => serde_json::json!({
218 "provider": provider,
219 "action": "added",
220 "timestamp": timestamp,
221 }),
222 CoreEvent::ProviderUpdated {
223 provider,
224 timestamp,
225 } => serde_json::json!({
226 "provider": provider,
227 "action": "updated",
228 "timestamp": timestamp,
229 }),
230 CoreEvent::ProviderRemoved {
231 provider,
232 timestamp,
233 } => serde_json::json!({
234 "provider": provider,
235 "action": "removed",
236 "timestamp": timestamp,
237 }),
238 CoreEvent::PipelineCacheInvalidated {
239 provider_id,
240 reason,
241 } => {
242 let mut json = serde_json::json!({
243 "reason": reason,
244 });
245 if let Some(pid) = provider_id {
246 json["providerId"] = serde_json::json!(pid);
247 }
248 json
249 }
250 CoreEvent::RunHistoryCacheInvalidated { pipeline_id } => {
251 let mut json = serde_json::json!({});
252 if let Some(pid) = pipeline_id {
253 json["pipeline_id"] = serde_json::json!(pid);
254 }
255 json
256 }
257 CoreEvent::VaultUnlocked => serde_json::json!({}),
258 }
259 }
260}
261
262#[async_trait]
263pub trait EventBus: Send + Sync {
264 async fn emit(&self, event: CoreEvent);
265
266 async fn emit_to(&self, target: &str, event: CoreEvent);
267}
268
269pub struct NoOpEventBus;
270
271#[async_trait]
272impl EventBus for NoOpEventBus {
273 async fn emit(&self, _event: CoreEvent) {}
274
275 async fn emit_to(&self, _target: &str, _event: CoreEvent) {}
276}
277
278#[cfg(test)]
279mod tests {
280 use super::*;
281
282 #[test]
283 fn test_event_names() {
284 assert_eq!(
285 CoreEvent::ProvidersChanged.event_name(),
286 "providers-changed"
287 );
288 assert_eq!(
289 CoreEvent::PipelineStatusChanged {
290 pipeline_ids: vec!["a".to_string()]
291 }
292 .event_name(),
293 "pipeline-status-changed"
294 );
295 }
296
297 #[test]
298 fn test_event_serialization() {
299 let event = CoreEvent::PipelineStatusChanged {
300 pipeline_ids: vec!["github__1__web".to_string(), "github__1__api".to_string()],
301 };
302 let json = serde_json::to_string(&event).unwrap();
303 assert!(json.contains("PipelineStatusChanged"));
304 assert!(json.contains("github__1__web"));
305 assert_eq!(
306 event.to_json_payload(),
307 serde_json::json!({ "pipeline_ids": ["github__1__web", "github__1__api"] })
308 );
309 }
310
311 #[test]
312 fn test_provider_health_alert_wire_format() {
313 let alert = |link| CoreEvent::ProviderHealthAlert {
314 alert_id: "alert-7".to_string(),
315 notification_key: "status-page:feed".to_string(),
316 provider_ids: vec![1, 2],
317 level: HealthLevel::MajorOutage,
318 title: "GitHub is down".to_string(),
319 body: "Investigating".to_string(),
320 link,
321 recovered: false,
322 };
323 let with_link = alert(Some(AlertLink {
324 url: "https://www.githubstatus.com/incidents/abc".to_string(),
325 kind: AlertLinkKind::Incident,
326 }));
327
328 assert_eq!(with_link.event_name(), "provider-health-alert");
329 assert_eq!(
330 with_link.to_json_payload(),
331 serde_json::json!({
332 "alert_id": "alert-7",
333 "notification_key": "status-page:feed",
334 "provider_ids": [1, 2],
335 "level": "major_outage",
336 "title": "GitHub is down",
337 "body": "Investigating",
338 "link": {
339 "url": "https://www.githubstatus.com/incidents/abc",
340 "kind": "incident"
341 },
342 "recovered": false,
343 })
344 );
345 assert_eq!(
346 alert(None).to_json_payload()["link"],
347 serde_json::Value::Null
348 );
349 }
350}