1use std::collections::HashMap;
4
5use chrono::{DateTime, Utc};
6use rust_decimal::Decimal;
7use serde::{Deserialize, Serialize};
8use uuid::Uuid;
9
10pub use ironflow_store::entities::LogStream;
11use ironflow_store::models::{
12 ApprovalRequirement, Assignee, ProviderAccountWindow, RunStatus, ScheduleMissReason, StepKind,
13};
14
15fn default_approval_count() -> u32 {
18 1
19}
20
21#[derive(Debug, Clone, Serialize, Deserialize)]
38#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
39pub struct RunCreatedEvent {
40 pub run_id: Uuid,
42 pub workflow_name: String,
44 pub at: DateTime<Utc>,
46}
47
48#[derive(Debug, Clone, Serialize, Deserialize)]
75#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
76pub struct RunStatusChangedEvent {
77 pub run_id: Uuid,
79 pub workflow_name: String,
81 pub from: RunStatus,
83 pub to: RunStatus,
85 pub error: Option<String>,
87 pub cost_usd: Decimal,
89 pub duration_ms: u64,
91 #[serde(default)]
93 pub labels: HashMap<String, String>,
94 pub at: DateTime<Utc>,
96}
97
98#[derive(Debug, Clone, Serialize, Deserialize)]
122#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
123pub struct RunFailedEvent {
124 pub run_id: Uuid,
126 pub workflow_name: String,
128 pub error: Option<String>,
130 pub cost_usd: Decimal,
132 pub duration_ms: u64,
134 #[serde(default)]
136 pub labels: HashMap<String, String>,
137 pub at: DateTime<Utc>,
139}
140
141#[derive(Debug, Clone, Serialize, Deserialize)]
162#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
163pub struct RunBudgetExceededEvent {
164 pub run_id: Uuid,
166 pub workflow_name: String,
168 pub limit_usd: Decimal,
170 pub spent_usd: Decimal,
172 pub step_budget_usd: Decimal,
174 pub at: DateTime<Utc>,
176}
177
178#[derive(Debug, Clone, Serialize, Deserialize)]
197#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
198pub struct RetryForcedEvent {
199 pub run_id: Uuid,
201 pub workflow_name: String,
203 pub original_version: String,
205 pub current_version: String,
207 pub at: DateTime<Utc>,
209}
210
211#[derive(Debug, Clone, Serialize, Deserialize)]
234#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
235pub struct StepCompletedEvent {
236 pub run_id: Uuid,
238 pub step_id: Uuid,
240 pub step_name: String,
242 #[cfg_attr(feature = "openapi", schema(value_type = String))]
244 pub kind: StepKind,
245 pub duration_ms: u64,
247 pub cost_usd: Decimal,
249 pub at: DateTime<Utc>,
251}
252
253#[derive(Debug, Clone, Serialize, Deserialize)]
274#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
275pub struct StepFailedEvent {
276 pub run_id: Uuid,
278 pub step_id: Uuid,
280 pub step_name: String,
282 #[cfg_attr(feature = "openapi", schema(value_type = String))]
284 pub kind: StepKind,
285 pub error: String,
287 pub at: DateTime<Utc>,
289}
290
291#[derive(Debug, Clone, Serialize, Deserialize)]
313#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
314pub struct ApprovalRequestedEvent {
315 pub run_id: Uuid,
317 pub step_id: Uuid,
319 pub message: String,
321 #[serde(default)]
324 pub requirement: Option<ApprovalRequirement>,
325 pub at: DateTime<Utc>,
327}
328
329#[derive(Debug, Clone, Serialize, Deserialize)]
356#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
357pub struct ApprovalGrantedEvent {
358 pub run_id: Uuid,
360 #[serde(default)]
362 pub step_id: Option<Uuid>,
363 pub approved_by: String,
365 #[serde(default)]
369 pub user_id: Option<Uuid>,
370 #[serde(default = "default_approval_count")]
372 pub approvals_received: u32,
373 #[serde(default = "default_approval_count")]
375 pub approvals_required: u32,
376 #[serde(default)]
378 pub requirement: Option<ApprovalRequirement>,
379 pub at: DateTime<Utc>,
381}
382
383#[derive(Debug, Clone, Serialize, Deserialize)]
404#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
405pub struct ApprovalRejectedEvent {
406 pub run_id: Uuid,
408 #[serde(default)]
410 pub step_id: Option<Uuid>,
411 pub rejected_by: String,
413 #[serde(default)]
417 pub user_id: Option<Uuid>,
418 #[serde(default)]
421 pub reason: Option<String>,
422 #[serde(default)]
424 pub requirement: Option<ApprovalRequirement>,
425 pub at: DateTime<Utc>,
427}
428
429#[derive(Debug, Clone, Serialize, Deserialize)]
457#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
458pub struct ApprovalEscalatedEvent {
459 pub run_id: Uuid,
461 pub step_id: Uuid,
463 pub step_name: String,
465 pub stage: u32,
467 pub policy: String,
469 pub action: String,
471 pub reason: String,
473 #[cfg_attr(feature = "openapi", schema(value_type = Option<String>))]
475 pub assignee: Option<Assignee>,
476 pub at: DateTime<Utc>,
478}
479
480#[derive(Debug, Clone, Serialize, Deserialize)]
501#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
502pub struct LogLineEvent {
503 #[serde(default)]
512 pub id: Uuid,
513 pub run_id: Uuid,
515 pub step_id: Uuid,
517 pub step_name: String,
519 pub stream: LogStream,
521 pub line: String,
523 pub at: DateTime<Utc>,
525}
526
527#[derive(Debug, Clone, Serialize, Deserialize)]
544#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
545pub struct UserSignedInEvent {
546 pub user_id: Uuid,
548 pub username: String,
550 pub at: DateTime<Utc>,
552}
553
554#[derive(Debug, Clone, Serialize, Deserialize)]
571#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
572pub struct UserSignedUpEvent {
573 pub user_id: Uuid,
575 pub username: String,
577 pub at: DateTime<Utc>,
579}
580
581#[derive(Debug, Clone, Serialize, Deserialize)]
598#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
599pub struct UserSignedOutEvent {
600 pub user_id: Uuid,
602 pub at: DateTime<Utc>,
604}
605
606#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
617#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
618#[serde(rename_all = "snake_case")]
619pub enum ProviderAccountChange {
620 Created,
622 Updated,
624 Deleted,
626 TokenReplaced,
628}
629
630impl ProviderAccountChange {
631 pub fn as_str(&self) -> &'static str {
633 match self {
634 Self::Created => "created",
635 Self::Updated => "updated",
636 Self::Deleted => "deleted",
637 Self::TokenReplaced => "token_replaced",
638 }
639 }
640}
641
642#[derive(Debug, Clone, Serialize, Deserialize)]
662#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
663pub struct ProviderAccountUpdatedEvent {
664 pub account_id: Uuid,
666 pub name: String,
668 pub change: ProviderAccountChange,
670 pub at: DateTime<Utc>,
672}
673
674#[derive(Debug, Clone, Serialize, Deserialize)]
692#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
693pub struct ProviderAccountUsageUpdatedEvent {
694 pub account_id: Uuid,
696 pub name: String,
698 pub windows: Vec<ProviderAccountWindow>,
700 pub at: DateTime<Utc>,
702}
703
704#[derive(Debug, Clone, Serialize, Deserialize)]
728#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
729pub struct SignalAwaitedEvent {
730 pub run_id: Uuid,
732 pub step_id: Uuid,
734 pub step_name: String,
736 pub name: String,
738 pub key: String,
740 pub deadline_at: DateTime<Utc>,
742 pub at: DateTime<Utc>,
744}
745
746#[derive(Debug, Clone, Serialize, Deserialize)]
770#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
771pub struct SignalReceivedEvent {
772 pub signal_id: Uuid,
774 pub name: String,
776 pub key: String,
778 pub resumed_runs: Vec<Uuid>,
780 #[serde(default, skip_serializing_if = "Option::is_none")]
783 pub sender_run_id: Option<Uuid>,
784 #[serde(default, skip_serializing_if = "Option::is_none")]
786 pub sender_step_id: Option<Uuid>,
787 pub at: DateTime<Utc>,
789}
790
791#[derive(Debug, Clone, Serialize, Deserialize)]
818#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
819pub struct ScheduleOccurrencesMissedEvent {
820 pub schedule_id: Uuid,
822 pub workflow_name: String,
824 pub reason: ScheduleMissReason,
826 pub count: u64,
828 pub first: DateTime<Utc>,
830 pub last: DateTime<Utc>,
832 pub at: DateTime<Utc>,
834}
835
836#[derive(Debug, Clone, Serialize, Deserialize)]
856#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
857pub struct WorkflowPausedEvent {
858 pub workflow_name: String,
860 #[serde(default)]
863 pub user_id: Option<Uuid>,
864 pub at: DateTime<Utc>,
866}
867
868#[derive(Debug, Clone, Serialize, Deserialize)]
888#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
889pub struct WorkflowResumedEvent {
890 pub workflow_name: String,
892 #[serde(default)]
895 pub user_id: Option<Uuid>,
896 pub at: DateTime<Utc>,
898}
899
900#[derive(Debug, Clone, Serialize, Deserialize)]
932#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
933#[serde(tag = "type", rename_all = "snake_case")]
934pub enum Event {
935 RunCreated(RunCreatedEvent),
938
939 RunStatusChanged(RunStatusChangedEvent),
941
942 RunFailed(RunFailedEvent),
948
949 RunBudgetExceeded(RunBudgetExceededEvent),
957
958 RetryForced(RetryForcedEvent),
965
966 StepCompleted(StepCompletedEvent),
969
970 StepFailed(StepFailedEvent),
972
973 ApprovalRequested(ApprovalRequestedEvent),
976
977 ApprovalGranted(ApprovalGrantedEvent),
979
980 ApprovalRejected(ApprovalRejectedEvent),
982
983 ApprovalEscalated(ApprovalEscalatedEvent),
985
986 LogLine(LogLineEvent),
992
993 UserSignedIn(UserSignedInEvent),
996
997 UserSignedUp(UserSignedUpEvent),
999
1000 UserSignedOut(UserSignedOutEvent),
1002
1003 #[serde(rename = "provider_account.updated")]
1006 ProviderAccountUpdated(ProviderAccountUpdatedEvent),
1007
1008 #[serde(rename = "provider_account.usage_updated")]
1010 ProviderAccountUsageUpdated(ProviderAccountUsageUpdatedEvent),
1011
1012 SignalAwaited(SignalAwaitedEvent),
1015
1016 SignalReceived(SignalReceivedEvent),
1018
1019 ScheduleOccurrencesMissed(ScheduleOccurrencesMissedEvent),
1024
1025 WorkflowPaused(WorkflowPausedEvent),
1028
1029 WorkflowResumed(WorkflowResumedEvent),
1031}
1032
1033impl Event {
1034 pub const RUN_CREATED: &'static str = "run_created";
1036 pub const RUN_STATUS_CHANGED: &'static str = "run_status_changed";
1038 pub const RUN_FAILED: &'static str = "run_failed";
1040 pub const RUN_BUDGET_EXCEEDED: &'static str = "run_budget_exceeded";
1042 pub const RETRY_FORCED: &'static str = "retry_forced";
1044 pub const STEP_COMPLETED: &'static str = "step_completed";
1046 pub const STEP_FAILED: &'static str = "step_failed";
1048 pub const APPROVAL_REQUESTED: &'static str = "approval_requested";
1050 pub const APPROVAL_GRANTED: &'static str = "approval_granted";
1052 pub const APPROVAL_REJECTED: &'static str = "approval_rejected";
1054 pub const APPROVAL_ESCALATED: &'static str = "approval_escalated";
1056 pub const LOG_LINE: &'static str = "log_line";
1058 pub const USER_SIGNED_IN: &'static str = "user_signed_in";
1060 pub const USER_SIGNED_UP: &'static str = "user_signed_up";
1062 pub const USER_SIGNED_OUT: &'static str = "user_signed_out";
1064 pub const PROVIDER_ACCOUNT_UPDATED: &'static str = "provider_account.updated";
1066 pub const PROVIDER_ACCOUNT_USAGE_UPDATED: &'static str = "provider_account.usage_updated";
1069 pub const SIGNAL_AWAITED: &'static str = "signal_awaited";
1071 pub const SIGNAL_RECEIVED: &'static str = "signal_received";
1073 pub const SCHEDULE_OCCURRENCES_MISSED: &'static str = "schedule_occurrences_missed";
1076 pub const WORKFLOW_PAUSED: &'static str = "workflow_paused";
1078 pub const WORKFLOW_RESUMED: &'static str = "workflow_resumed";
1080
1081 pub const ALL: &'static [&'static str] = &[
1097 Self::RUN_CREATED,
1098 Self::RUN_STATUS_CHANGED,
1099 Self::RUN_FAILED,
1100 Self::RUN_BUDGET_EXCEEDED,
1101 Self::STEP_COMPLETED,
1102 Self::STEP_FAILED,
1103 Self::APPROVAL_REQUESTED,
1104 Self::APPROVAL_GRANTED,
1105 Self::APPROVAL_REJECTED,
1106 Self::APPROVAL_ESCALATED,
1107 Self::LOG_LINE,
1108 Self::USER_SIGNED_IN,
1109 Self::USER_SIGNED_UP,
1110 Self::USER_SIGNED_OUT,
1111 Self::RETRY_FORCED,
1112 Self::PROVIDER_ACCOUNT_UPDATED,
1113 Self::PROVIDER_ACCOUNT_USAGE_UPDATED,
1114 Self::SIGNAL_AWAITED,
1115 Self::SIGNAL_RECEIVED,
1116 Self::SCHEDULE_OCCURRENCES_MISSED,
1117 Self::WORKFLOW_PAUSED,
1118 Self::WORKFLOW_RESUMED,
1119 ];
1120
1121 #[deny(unreachable_patterns)]
1140 pub fn event_type(&self) -> &'static str {
1141 match self {
1142 Event::RunCreated(_) => Self::RUN_CREATED,
1143 Event::RunStatusChanged(_) => Self::RUN_STATUS_CHANGED,
1144 Event::RunFailed(_) => Self::RUN_FAILED,
1145 Event::RunBudgetExceeded(_) => Self::RUN_BUDGET_EXCEEDED,
1146 Event::RetryForced(_) => Self::RETRY_FORCED,
1147 Event::StepCompleted(_) => Self::STEP_COMPLETED,
1148 Event::StepFailed(_) => Self::STEP_FAILED,
1149 Event::ApprovalRequested(_) => Self::APPROVAL_REQUESTED,
1150 Event::ApprovalGranted(_) => Self::APPROVAL_GRANTED,
1151 Event::ApprovalRejected(_) => Self::APPROVAL_REJECTED,
1152 Event::ApprovalEscalated(_) => Self::APPROVAL_ESCALATED,
1153 Event::LogLine(_) => Self::LOG_LINE,
1154 Event::UserSignedIn(_) => Self::USER_SIGNED_IN,
1155 Event::UserSignedUp(_) => Self::USER_SIGNED_UP,
1156 Event::UserSignedOut(_) => Self::USER_SIGNED_OUT,
1157 Event::ProviderAccountUpdated(_) => Self::PROVIDER_ACCOUNT_UPDATED,
1158 Event::ProviderAccountUsageUpdated(_) => Self::PROVIDER_ACCOUNT_USAGE_UPDATED,
1159 Event::SignalAwaited(_) => Self::SIGNAL_AWAITED,
1160 Event::SignalReceived(_) => Self::SIGNAL_RECEIVED,
1161 Event::ScheduleOccurrencesMissed(_) => Self::SCHEDULE_OCCURRENCES_MISSED,
1162 Event::WorkflowPaused(_) => Self::WORKFLOW_PAUSED,
1163 Event::WorkflowResumed(_) => Self::WORKFLOW_RESUMED,
1164 }
1165 }
1166
1167 #[deny(unreachable_patterns)]
1191 pub fn run_id(&self) -> Option<Uuid> {
1192 match self {
1193 Event::RunCreated(e) => Some(e.run_id),
1194 Event::RunStatusChanged(e) => Some(e.run_id),
1195 Event::RunFailed(e) => Some(e.run_id),
1196 Event::RunBudgetExceeded(e) => Some(e.run_id),
1197 Event::RetryForced(e) => Some(e.run_id),
1198 Event::StepCompleted(e) => Some(e.run_id),
1199 Event::StepFailed(e) => Some(e.run_id),
1200 Event::ApprovalRequested(e) => Some(e.run_id),
1201 Event::ApprovalGranted(e) => Some(e.run_id),
1202 Event::ApprovalRejected(e) => Some(e.run_id),
1203 Event::ApprovalEscalated(e) => Some(e.run_id),
1204 Event::LogLine(e) => Some(e.run_id),
1205 Event::SignalAwaited(e) => Some(e.run_id),
1206 Event::SignalReceived(e) => e.sender_run_id,
1207 Event::UserSignedIn(_)
1208 | Event::UserSignedUp(_)
1209 | Event::UserSignedOut(_)
1210 | Event::ProviderAccountUpdated(_)
1211 | Event::ProviderAccountUsageUpdated(_)
1212 | Event::ScheduleOccurrencesMissed(_)
1213 | Event::WorkflowPaused(_)
1214 | Event::WorkflowResumed(_) => None,
1215 }
1216 }
1217
1218 #[deny(unreachable_patterns)]
1249 pub fn step_id(&self) -> Option<Uuid> {
1250 match self {
1251 Event::StepCompleted(e) => Some(e.step_id),
1252 Event::StepFailed(e) => Some(e.step_id),
1253 Event::ApprovalRequested(e) => Some(e.step_id),
1254 Event::ApprovalEscalated(e) => Some(e.step_id),
1255 Event::ApprovalGranted(e) => e.step_id,
1256 Event::ApprovalRejected(e) => e.step_id,
1257 Event::SignalAwaited(e) => Some(e.step_id),
1258 Event::SignalReceived(e) => e.sender_step_id,
1259 Event::RunCreated(_)
1260 | Event::RunStatusChanged(_)
1261 | Event::RunFailed(_)
1262 | Event::RunBudgetExceeded(_)
1263 | Event::RetryForced(_)
1264 | Event::LogLine(_)
1265 | Event::UserSignedIn(_)
1266 | Event::UserSignedUp(_)
1267 | Event::UserSignedOut(_)
1268 | Event::ProviderAccountUpdated(_)
1269 | Event::ProviderAccountUsageUpdated(_)
1270 | Event::ScheduleOccurrencesMissed(_)
1271 | Event::WorkflowPaused(_)
1272 | Event::WorkflowResumed(_) => None,
1273 }
1274 }
1275
1276 #[deny(unreachable_patterns)]
1304 pub fn user_id(&self) -> Option<Uuid> {
1305 match self {
1306 Event::UserSignedIn(e) => Some(e.user_id),
1307 Event::UserSignedUp(e) => Some(e.user_id),
1308 Event::UserSignedOut(e) => Some(e.user_id),
1309 Event::ApprovalGranted(e) => e.user_id,
1310 Event::ApprovalRejected(e) => e.user_id,
1311 Event::WorkflowPaused(e) => e.user_id,
1312 Event::WorkflowResumed(e) => e.user_id,
1313 Event::RunCreated(_)
1314 | Event::RunStatusChanged(_)
1315 | Event::RunFailed(_)
1316 | Event::RunBudgetExceeded(_)
1317 | Event::RetryForced(_)
1318 | Event::StepCompleted(_)
1319 | Event::StepFailed(_)
1320 | Event::ApprovalRequested(_)
1321 | Event::ApprovalEscalated(_)
1322 | Event::LogLine(_)
1323 | Event::ProviderAccountUpdated(_)
1324 | Event::ProviderAccountUsageUpdated(_)
1325 | Event::SignalAwaited(_)
1326 | Event::SignalReceived(_)
1327 | Event::ScheduleOccurrencesMissed(_) => None,
1328 }
1329 }
1330}
1331
1332#[cfg(test)]
1333mod tests {
1334 use serde_json::{from_value, json};
1335
1336 use super::*;
1337
1338 #[test]
1339 fn run_status_changed_serde_roundtrip() {
1340 let event = Event::RunStatusChanged(RunStatusChangedEvent {
1341 run_id: Uuid::now_v7(),
1342 workflow_name: "deploy".to_string(),
1343 from: RunStatus::Running,
1344 to: RunStatus::Completed,
1345 error: None,
1346 cost_usd: Decimal::new(42, 2),
1347 duration_ms: 5000,
1348 labels: HashMap::new(),
1349 at: Utc::now(),
1350 });
1351
1352 let json = serde_json::to_string(&event).expect("serialize");
1353 let back: Event = serde_json::from_str(&json).expect("deserialize");
1354
1355 assert_eq!(back.event_type(), "run_status_changed");
1356 assert!(json.contains("\"type\":\"run_status_changed\""));
1357 }
1358
1359 #[test]
1360 fn run_failed_serde_roundtrip() {
1361 let event = Event::RunFailed(RunFailedEvent {
1362 run_id: Uuid::now_v7(),
1363 workflow_name: "deploy".to_string(),
1364 error: Some("step crashed".to_string()),
1365 cost_usd: Decimal::new(10, 2),
1366 duration_ms: 3000,
1367 labels: HashMap::new(),
1368 at: Utc::now(),
1369 });
1370
1371 let json = serde_json::to_string(&event).expect("serialize");
1372 let back: Event = serde_json::from_str(&json).expect("deserialize");
1373
1374 assert_eq!(back.event_type(), "run_failed");
1375 assert!(json.contains("\"type\":\"run_failed\""));
1376 assert!(json.contains("step crashed"));
1377 }
1378
1379 #[test]
1380 fn run_budget_exceeded_serde_roundtrip() {
1381 let event = Event::RunBudgetExceeded(RunBudgetExceededEvent {
1382 run_id: Uuid::now_v7(),
1383 workflow_name: "deploy".to_string(),
1384 limit_usd: Decimal::new(200, 2),
1385 spent_usd: Decimal::new(180, 2),
1386 step_budget_usd: Decimal::new(50, 2),
1387 at: Utc::now(),
1388 });
1389
1390 let json = serde_json::to_string(&event).expect("serialize");
1391 let back: Event = serde_json::from_str(&json).expect("deserialize");
1392
1393 assert_eq!(back.event_type(), "run_budget_exceeded");
1394 assert!(json.contains("\"type\":\"run_budget_exceeded\""));
1395 assert!(json.contains("limit_usd"));
1396 assert!(json.contains("step_budget_usd"));
1397 }
1398
1399 #[test]
1400 fn all_contains_run_budget_exceeded() {
1401 assert!(Event::ALL.contains(&Event::RUN_BUDGET_EXCEEDED));
1402 }
1403
1404 #[test]
1405 fn user_signed_in_serde_roundtrip() {
1406 let event = Event::UserSignedIn(UserSignedInEvent {
1407 user_id: Uuid::now_v7(),
1408 username: "alice".to_string(),
1409 at: Utc::now(),
1410 });
1411
1412 let json = serde_json::to_string(&event).expect("serialize");
1413 let back: Event = serde_json::from_str(&json).expect("deserialize");
1414
1415 assert_eq!(back.event_type(), "user_signed_in");
1416 assert!(json.contains("alice"));
1417 }
1418
1419 #[test]
1420 fn step_failed_serde_roundtrip() {
1421 let event = Event::StepFailed(StepFailedEvent {
1422 run_id: Uuid::now_v7(),
1423 step_id: Uuid::now_v7(),
1424 step_name: "build".to_string(),
1425 kind: StepKind::Shell,
1426 error: "exit code 1".to_string(),
1427 at: Utc::now(),
1428 });
1429
1430 let json = serde_json::to_string(&event).expect("serialize");
1431 let back: Event = serde_json::from_str(&json).expect("deserialize");
1432
1433 assert_eq!(back.event_type(), "step_failed");
1434 }
1435
1436 #[test]
1437 fn legacy_approval_granted_defaults_to_a_single_vote() {
1438 let raw = r#"{"type":"approval_granted","run_id":"01890000-0000-7000-8000-000000000000","approved_by":"alice","at":"2026-01-01T00:00:00Z"}"#;
1439 let event: Event = serde_json::from_str(raw).expect("deserialize");
1440 let Event::ApprovalGranted(event) = event else {
1441 panic!("expected approval_granted");
1442 };
1443
1444 assert_eq!(event.step_id, None);
1445 assert_eq!(event.approvals_received, 1);
1446 assert_eq!(event.approvals_required, 1);
1447 assert!(event.requirement.is_none());
1448 }
1449
1450 #[test]
1451 fn legacy_approval_requested_and_rejected_have_no_requirement() {
1452 let raw = r#"{"type":"approval_requested","run_id":"01890000-0000-7000-8000-000000000000","step_id":"01890000-0000-7000-8000-000000000001","message":"ok?","at":"2026-01-01T00:00:00Z"}"#;
1453 let requested: Event = serde_json::from_str(raw).expect("deserialize");
1454 let Event::ApprovalRequested(requested) = requested else {
1455 panic!("expected approval_requested");
1456 };
1457 assert!(requested.requirement.is_none());
1458
1459 let raw = r#"{"type":"approval_rejected","run_id":"01890000-0000-7000-8000-000000000000","rejected_by":"bob","at":"2026-01-01T00:00:00Z"}"#;
1460 let rejected: Event = serde_json::from_str(raw).expect("deserialize");
1461 let Event::ApprovalRejected(rejected) = rejected else {
1462 panic!("expected approval_rejected");
1463 };
1464 assert_eq!(rejected.step_id, None);
1465 assert!(rejected.requirement.is_none());
1466 }
1467
1468 #[test]
1469 fn approval_granted_roundtrips_the_vote_counts() {
1470 let requirement = ApprovalRequirement {
1471 reason: Some("amount > 10k".to_string()),
1472 required_approvers: 2,
1473 approver_groups: vec!["finance".to_string()],
1474 };
1475 let event = Event::ApprovalGranted(ApprovalGrantedEvent {
1476 run_id: Uuid::now_v7(),
1477 step_id: Some(Uuid::now_v7()),
1478 approved_by: "alice".to_string(),
1479 user_id: None,
1480 approvals_received: 1,
1481 approvals_required: 2,
1482 requirement: Some(requirement.clone()),
1483 at: Utc::now(),
1484 });
1485
1486 let json = serde_json::to_string(&event).expect("serialize");
1487 let back: Event = serde_json::from_str(&json).expect("deserialize");
1488 let Event::ApprovalGranted(back) = back else {
1489 panic!("expected approval_granted");
1490 };
1491 assert_eq!(back.approvals_received, 1);
1492 assert_eq!(back.approvals_required, 2);
1493 assert_eq!(back.requirement, Some(requirement));
1494 }
1495
1496 #[test]
1497 fn approval_requested_serde_roundtrip() {
1498 let event = Event::ApprovalRequested(ApprovalRequestedEvent {
1499 run_id: Uuid::now_v7(),
1500 step_id: Uuid::now_v7(),
1501 message: "Deploy to prod?".to_string(),
1502 requirement: None,
1503 at: Utc::now(),
1504 });
1505
1506 let json = serde_json::to_string(&event).expect("serialize");
1507 assert!(json.contains("approval_requested"));
1508 }
1509
1510 #[test]
1511 fn log_line_serde_roundtrip() {
1512 let event = Event::LogLine(LogLineEvent {
1513 id: Uuid::now_v7(),
1514 run_id: Uuid::now_v7(),
1515 step_id: Uuid::now_v7(),
1516 step_name: "build".to_string(),
1517 stream: LogStream::Stdout,
1518 line: "Compiling ironflow v0.1.0".to_string(),
1519 at: Utc::now(),
1520 });
1521
1522 let json = serde_json::to_string(&event).expect("serialize");
1523 let back: Event = serde_json::from_str(&json).expect("deserialize");
1524
1525 assert_eq!(back.event_type(), "log_line");
1526 assert!(json.contains("\"type\":\"log_line\""));
1527 assert!(json.contains("Compiling ironflow"));
1528 }
1529
1530 #[test]
1535 fn legacy_flat_json_deserializes_into_typed_payload() {
1536 let run_id: Uuid = "01890000-0000-7000-8000-000000000000"
1537 .parse()
1538 .expect("valid uuid");
1539
1540 let raw = r#"{"type":"run_created","run_id":"01890000-0000-7000-8000-000000000000","workflow_name":"deploy","at":"2026-01-01T00:00:00Z"}"#;
1541 let event: Event = serde_json::from_str(raw).expect("legacy payload must deserialize");
1542 match event {
1543 Event::RunCreated(e) => {
1544 assert_eq!(e.run_id, run_id);
1545 assert_eq!(e.workflow_name, "deploy");
1546 }
1547 other => panic!("expected RunCreated, got {other:?}"),
1548 }
1549
1550 let raw = r#"{"type":"run_status_changed","run_id":"01890000-0000-7000-8000-000000000000","workflow_name":"deploy","from":"running","to":"completed","error":null,"cost_usd":0.5,"duration_ms":5000,"labels":{"env":"prod"},"at":"2026-01-01T00:00:00Z"}"#;
1551 let event: Event = serde_json::from_str(raw).expect("legacy payload must deserialize");
1552 match event {
1553 Event::RunStatusChanged(e) => {
1554 assert_eq!(e.from, RunStatus::Running);
1555 assert_eq!(e.to, RunStatus::Completed);
1556 assert_eq!(e.cost_usd, Decimal::new(5, 1));
1557 assert_eq!(e.duration_ms, 5000);
1558 assert_eq!(e.labels.get("env").map(String::as_str), Some("prod"));
1559 }
1560 other => panic!("expected RunStatusChanged, got {other:?}"),
1561 }
1562
1563 let raw = r#"{"type":"run_status_changed","run_id":"01890000-0000-7000-8000-000000000000","workflow_name":"deploy","from":"running","to":"failed","error":"boom","cost_usd":0,"duration_ms":0,"at":"2026-01-01T00:00:00Z"}"#;
1565 let event: Event = serde_json::from_str(raw).expect("missing labels must default");
1566 match event {
1567 Event::RunStatusChanged(e) => {
1568 assert!(e.labels.is_empty());
1569 assert_eq!(e.error.as_deref(), Some("boom"));
1570 }
1571 other => panic!("expected RunStatusChanged, got {other:?}"),
1572 }
1573
1574 let raw = r#"{"type":"run_failed","run_id":"01890000-0000-7000-8000-000000000000","workflow_name":"deploy","error":"boom","cost_usd":0.25,"duration_ms":3000,"at":"2026-01-01T00:00:00Z"}"#;
1575 let event: Event = serde_json::from_str(raw).expect("legacy payload must deserialize");
1576 match event {
1577 Event::RunFailed(e) => {
1578 assert_eq!(e.error.as_deref(), Some("boom"));
1579 assert!(e.labels.is_empty());
1580 }
1581 other => panic!("expected RunFailed, got {other:?}"),
1582 }
1583
1584 let raw = r#"{"type":"step_failed","run_id":"01890000-0000-7000-8000-000000000000","step_id":"01890000-0000-7000-8000-000000000001","step_name":"build","kind":"shell","error":"exit code 1","at":"2026-01-01T00:00:00Z"}"#;
1585 let event: Event = serde_json::from_str(raw).expect("legacy payload must deserialize");
1586 match event {
1587 Event::StepFailed(e) => {
1588 assert_eq!(e.kind, StepKind::Shell);
1589 assert_eq!(e.error, "exit code 1");
1590 }
1591 other => panic!("expected StepFailed, got {other:?}"),
1592 }
1593
1594 let raw = r#"{"type":"log_line","run_id":"01890000-0000-7000-8000-000000000000","step_id":"01890000-0000-7000-8000-000000000001","step_name":"build","stream":"stdout","line":"hello","at":"2026-01-01T00:00:00Z"}"#;
1595 let event: Event = serde_json::from_str(raw).expect("legacy payload must deserialize");
1596 match event {
1597 Event::LogLine(e) => {
1598 assert_eq!(e.stream, LogStream::Stdout);
1599 assert_eq!(e.line, "hello");
1600 assert_eq!(e.id, Uuid::nil());
1603 }
1604 other => panic!("expected LogLine, got {other:?}"),
1605 }
1606
1607 let raw = r#"{"type":"user_signed_in","user_id":"01890000-0000-7000-8000-000000000000","username":"alice","at":"2026-01-01T00:00:00Z"}"#;
1608 let event: Event = serde_json::from_str(raw).expect("legacy payload must deserialize");
1609 match event {
1610 Event::UserSignedIn(e) => assert_eq!(e.username, "alice"),
1611 other => panic!("expected UserSignedIn, got {other:?}"),
1612 }
1613 }
1614
1615 #[test]
1618 fn serialized_event_is_flat_with_type_tag() {
1619 let run_id = Uuid::now_v7();
1620 let event = Event::RunCreated(RunCreatedEvent {
1621 run_id,
1622 workflow_name: "deploy".to_string(),
1623 at: Utc::now(),
1624 });
1625
1626 let value: serde_json::Value = serde_json::to_value(&event).expect("serialize");
1627 let object = value.as_object().expect("event serializes to an object");
1628
1629 assert_eq!(
1630 object.get("type").and_then(|v| v.as_str()),
1631 Some("run_created")
1632 );
1633 assert_eq!(
1634 object.get("workflow_name").and_then(|v| v.as_str()),
1635 Some("deploy")
1636 );
1637 assert_eq!(
1638 object.get("run_id").and_then(|v| v.as_str()),
1639 Some(run_id.to_string().as_str())
1640 );
1641 assert!(object.contains_key("at"));
1642 assert_eq!(object.len(), 4, "no nesting: {object:?}");
1643 assert!(!object.contains_key("RunCreated"));
1644 }
1645
1646 #[test]
1647 fn run_id_returns_some_for_run_events() {
1648 let run_id = Uuid::now_v7();
1649 let now = Utc::now();
1650
1651 let events = vec![
1652 Event::RunCreated(RunCreatedEvent {
1653 run_id,
1654 workflow_name: "w".to_string(),
1655 at: now,
1656 }),
1657 Event::RunStatusChanged(RunStatusChangedEvent {
1658 run_id,
1659 workflow_name: "w".to_string(),
1660 from: RunStatus::Pending,
1661 to: RunStatus::Running,
1662 error: None,
1663 cost_usd: Decimal::ZERO,
1664 duration_ms: 0,
1665 labels: HashMap::new(),
1666 at: now,
1667 }),
1668 Event::RunFailed(RunFailedEvent {
1669 run_id,
1670 workflow_name: "w".to_string(),
1671 error: None,
1672 cost_usd: Decimal::ZERO,
1673 duration_ms: 0,
1674 labels: HashMap::new(),
1675 at: now,
1676 }),
1677 Event::RunBudgetExceeded(RunBudgetExceededEvent {
1678 run_id,
1679 workflow_name: "w".to_string(),
1680 limit_usd: Decimal::ZERO,
1681 spent_usd: Decimal::ZERO,
1682 step_budget_usd: Decimal::ZERO,
1683 at: now,
1684 }),
1685 Event::RetryForced(RetryForcedEvent {
1686 run_id,
1687 workflow_name: "w".to_string(),
1688 original_version: "1".to_string(),
1689 current_version: "2".to_string(),
1690 at: now,
1691 }),
1692 Event::StepCompleted(StepCompletedEvent {
1693 run_id,
1694 step_id: Uuid::now_v7(),
1695 step_name: "s".to_string(),
1696 kind: StepKind::Shell,
1697 duration_ms: 0,
1698 cost_usd: Decimal::ZERO,
1699 at: now,
1700 }),
1701 Event::StepFailed(StepFailedEvent {
1702 run_id,
1703 step_id: Uuid::now_v7(),
1704 step_name: "s".to_string(),
1705 kind: StepKind::Shell,
1706 error: "e".to_string(),
1707 at: now,
1708 }),
1709 Event::ApprovalRequested(ApprovalRequestedEvent {
1710 run_id,
1711 step_id: Uuid::now_v7(),
1712 message: "ok?".to_string(),
1713 requirement: None,
1714 at: now,
1715 }),
1716 Event::ApprovalGranted(ApprovalGrantedEvent {
1717 run_id,
1718 step_id: None,
1719 approved_by: "alice".to_string(),
1720 user_id: None,
1721 approvals_received: 1,
1722 approvals_required: 1,
1723 requirement: None,
1724 at: now,
1725 }),
1726 Event::ApprovalRejected(ApprovalRejectedEvent {
1727 run_id,
1728 step_id: None,
1729 rejected_by: "bob".to_string(),
1730 user_id: None,
1731 reason: None,
1732 requirement: None,
1733 at: now,
1734 }),
1735 Event::LogLine(LogLineEvent {
1736 id: Uuid::now_v7(),
1737 run_id,
1738 step_id: Uuid::now_v7(),
1739 step_name: "s".to_string(),
1740 stream: LogStream::Stdout,
1741 line: "l".to_string(),
1742 at: now,
1743 }),
1744 ];
1745
1746 for event in &events {
1747 assert_eq!(
1748 event.run_id(),
1749 Some(run_id),
1750 "{} should carry a run_id",
1751 event.event_type()
1752 );
1753 }
1754 }
1755
1756 #[test]
1757 fn run_id_returns_none_for_auth_events() {
1758 let user_id = Uuid::now_v7();
1759 let now = Utc::now();
1760
1761 let events = vec![
1762 Event::UserSignedIn(UserSignedInEvent {
1763 user_id,
1764 username: "alice".to_string(),
1765 at: now,
1766 }),
1767 Event::UserSignedUp(UserSignedUpEvent {
1768 user_id,
1769 username: "alice".to_string(),
1770 at: now,
1771 }),
1772 Event::UserSignedOut(UserSignedOutEvent { user_id, at: now }),
1773 ];
1774
1775 for event in &events {
1776 assert_eq!(event.run_id(), None, "{} has no run", event.event_type());
1777 }
1778 }
1779
1780 #[test]
1781 fn step_id_returns_some_only_for_step_events() {
1782 let step_id = Uuid::now_v7();
1783 let run_id = Uuid::now_v7();
1784 let now = Utc::now();
1785
1786 let with_step = vec![
1787 Event::StepCompleted(StepCompletedEvent {
1788 run_id,
1789 step_id,
1790 step_name: "s".to_string(),
1791 kind: StepKind::Shell,
1792 duration_ms: 0,
1793 cost_usd: Decimal::ZERO,
1794 at: now,
1795 }),
1796 Event::StepFailed(StepFailedEvent {
1797 run_id,
1798 step_id,
1799 step_name: "s".to_string(),
1800 kind: StepKind::Shell,
1801 error: "e".to_string(),
1802 at: now,
1803 }),
1804 Event::ApprovalRequested(ApprovalRequestedEvent {
1805 run_id,
1806 step_id,
1807 message: "ok?".to_string(),
1808 requirement: None,
1809 at: now,
1810 }),
1811 Event::ApprovalGranted(ApprovalGrantedEvent {
1812 run_id,
1813 step_id: Some(step_id),
1814 approved_by: "alice".to_string(),
1815 user_id: None,
1816 approvals_received: 1,
1817 approvals_required: 2,
1818 requirement: None,
1819 at: now,
1820 }),
1821 Event::ApprovalRejected(ApprovalRejectedEvent {
1822 run_id,
1823 step_id: Some(step_id),
1824 rejected_by: "bob".to_string(),
1825 user_id: None,
1826 reason: None,
1827 requirement: None,
1828 at: now,
1829 }),
1830 ];
1831
1832 for event in &with_step {
1833 assert_eq!(
1834 event.step_id(),
1835 Some(step_id),
1836 "{} should carry a step_id",
1837 event.event_type()
1838 );
1839 }
1840
1841 let without_step = vec![
1842 Event::RunCreated(RunCreatedEvent {
1843 run_id,
1844 workflow_name: "w".to_string(),
1845 at: now,
1846 }),
1847 Event::ApprovalGranted(ApprovalGrantedEvent {
1849 run_id,
1850 step_id: None,
1851 approved_by: "alice".to_string(),
1852 user_id: None,
1853 approvals_received: 1,
1854 approvals_required: 1,
1855 requirement: None,
1856 at: now,
1857 }),
1858 Event::LogLine(LogLineEvent {
1861 id: Uuid::now_v7(),
1862 run_id,
1863 step_id,
1864 step_name: "s".to_string(),
1865 stream: LogStream::Stdout,
1866 line: "l".to_string(),
1867 at: now,
1868 }),
1869 Event::UserSignedOut(UserSignedOutEvent {
1870 user_id: Uuid::now_v7(),
1871 at: now,
1872 }),
1873 ];
1874
1875 for event in &without_step {
1876 assert_eq!(
1877 event.step_id(),
1878 None,
1879 "{} should not carry a step_id",
1880 event.event_type()
1881 );
1882 }
1883 }
1884
1885 #[test]
1886 fn user_id_returns_the_actor_of_auth_events_decisions_and_pauses() {
1887 let user_id = Uuid::now_v7();
1888 let run_id = Uuid::now_v7();
1889 let now = Utc::now();
1890
1891 let auth = vec![
1892 Event::UserSignedIn(UserSignedInEvent {
1893 user_id,
1894 username: "alice".to_string(),
1895 at: now,
1896 }),
1897 Event::UserSignedUp(UserSignedUpEvent {
1898 user_id,
1899 username: "alice".to_string(),
1900 at: now,
1901 }),
1902 Event::UserSignedOut(UserSignedOutEvent { user_id, at: now }),
1903 Event::ApprovalGranted(ApprovalGrantedEvent {
1904 run_id,
1905 step_id: None,
1906 approved_by: "alice".to_string(),
1907 user_id: Some(user_id),
1908 approvals_received: 1,
1909 approvals_required: 1,
1910 requirement: None,
1911 at: now,
1912 }),
1913 Event::ApprovalRejected(ApprovalRejectedEvent {
1914 run_id,
1915 step_id: None,
1916 rejected_by: "bob".to_string(),
1917 user_id: Some(user_id),
1918 reason: None,
1919 requirement: None,
1920 at: now,
1921 }),
1922 Event::WorkflowPaused(WorkflowPausedEvent {
1923 workflow_name: "w".to_string(),
1924 user_id: Some(user_id),
1925 at: now,
1926 }),
1927 Event::WorkflowResumed(WorkflowResumedEvent {
1928 workflow_name: "w".to_string(),
1929 user_id: Some(user_id),
1930 at: now,
1931 }),
1932 ];
1933
1934 for event in &auth {
1935 assert_eq!(
1936 event.user_id(),
1937 Some(user_id),
1938 "{} should carry a user_id",
1939 event.event_type()
1940 );
1941 }
1942
1943 let non_auth = vec![
1944 Event::RunCreated(RunCreatedEvent {
1945 run_id,
1946 workflow_name: "w".to_string(),
1947 at: now,
1948 }),
1949 Event::StepFailed(StepFailedEvent {
1950 run_id,
1951 step_id: Uuid::now_v7(),
1952 step_name: "s".to_string(),
1953 kind: StepKind::Shell,
1954 error: "e".to_string(),
1955 at: now,
1956 }),
1957 Event::ApprovalRejected(ApprovalRejectedEvent {
1959 run_id,
1960 step_id: None,
1961 rejected_by: "system:timeout".to_string(),
1962 user_id: None,
1963 reason: None,
1964 requirement: None,
1965 at: now,
1966 }),
1967 Event::WorkflowPaused(WorkflowPausedEvent {
1968 workflow_name: "w".to_string(),
1969 user_id: None,
1970 at: now,
1971 }),
1972 ];
1973
1974 for event in &non_auth {
1975 assert_eq!(
1976 event.user_id(),
1977 None,
1978 "{} should not carry a user_id",
1979 event.event_type()
1980 );
1981 }
1982 }
1983
1984 #[test]
1985 fn event_type_all_variants() {
1986 let id = Uuid::now_v7();
1987 let now = Utc::now();
1988
1989 let cases: Vec<(Event, &str)> = vec![
1990 (
1991 Event::RunCreated(RunCreatedEvent {
1992 run_id: id,
1993 workflow_name: "w".to_string(),
1994 at: now,
1995 }),
1996 "run_created",
1997 ),
1998 (
1999 Event::RunStatusChanged(RunStatusChangedEvent {
2000 run_id: id,
2001 workflow_name: "w".to_string(),
2002 from: RunStatus::Pending,
2003 to: RunStatus::Running,
2004 error: None,
2005 cost_usd: Decimal::ZERO,
2006 duration_ms: 0,
2007 labels: HashMap::new(),
2008 at: now,
2009 }),
2010 "run_status_changed",
2011 ),
2012 (
2013 Event::RunFailed(RunFailedEvent {
2014 run_id: id,
2015 workflow_name: "w".to_string(),
2016 error: Some("boom".to_string()),
2017 cost_usd: Decimal::ZERO,
2018 duration_ms: 0,
2019 labels: HashMap::new(),
2020 at: now,
2021 }),
2022 "run_failed",
2023 ),
2024 (
2025 Event::RunBudgetExceeded(RunBudgetExceededEvent {
2026 run_id: id,
2027 workflow_name: "w".to_string(),
2028 limit_usd: Decimal::new(200, 2),
2029 spent_usd: Decimal::new(180, 2),
2030 step_budget_usd: Decimal::new(50, 2),
2031 at: now,
2032 }),
2033 "run_budget_exceeded",
2034 ),
2035 (
2036 Event::RetryForced(RetryForcedEvent {
2037 run_id: id,
2038 workflow_name: "w".to_string(),
2039 original_version: "1".to_string(),
2040 current_version: "2".to_string(),
2041 at: now,
2042 }),
2043 "retry_forced",
2044 ),
2045 (
2046 Event::StepCompleted(StepCompletedEvent {
2047 run_id: id,
2048 step_id: id,
2049 step_name: "s".to_string(),
2050 kind: StepKind::Shell,
2051 duration_ms: 0,
2052 cost_usd: Decimal::ZERO,
2053 at: now,
2054 }),
2055 "step_completed",
2056 ),
2057 (
2058 Event::StepFailed(StepFailedEvent {
2059 run_id: id,
2060 step_id: id,
2061 step_name: "s".to_string(),
2062 kind: StepKind::Shell,
2063 error: "err".to_string(),
2064 at: now,
2065 }),
2066 "step_failed",
2067 ),
2068 (
2069 Event::ApprovalRequested(ApprovalRequestedEvent {
2070 run_id: id,
2071 step_id: id,
2072 message: "ok?".to_string(),
2073 requirement: None,
2074 at: now,
2075 }),
2076 "approval_requested",
2077 ),
2078 (
2079 Event::ApprovalGranted(ApprovalGrantedEvent {
2080 run_id: id,
2081 step_id: Some(id),
2082 approved_by: "alice".to_string(),
2083 user_id: None,
2084 approvals_received: 1,
2085 approvals_required: 1,
2086 requirement: None,
2087 at: now,
2088 }),
2089 "approval_granted",
2090 ),
2091 (
2092 Event::ApprovalRejected(ApprovalRejectedEvent {
2093 run_id: id,
2094 step_id: Some(id),
2095 rejected_by: "bob".to_string(),
2096 user_id: None,
2097 reason: None,
2098 requirement: None,
2099 at: now,
2100 }),
2101 "approval_rejected",
2102 ),
2103 (
2104 Event::ApprovalEscalated(ApprovalEscalatedEvent {
2105 run_id: id,
2106 step_id: id,
2107 step_name: "prod-gate".to_string(),
2108 stage: 0,
2109 policy: "auto_reject".to_string(),
2110 action: "rejected".to_string(),
2111 reason: "approval deadline of 3600s expired".to_string(),
2112 assignee: None,
2113 at: now,
2114 }),
2115 "approval_escalated",
2116 ),
2117 (
2118 Event::LogLine(LogLineEvent {
2119 id,
2120 run_id: id,
2121 step_id: id,
2122 step_name: "build".to_string(),
2123 stream: LogStream::Stdout,
2124 line: "Compiling ironflow v0.1.0".to_string(),
2125 at: now,
2126 }),
2127 "log_line",
2128 ),
2129 (
2130 Event::UserSignedIn(UserSignedInEvent {
2131 user_id: id,
2132 username: "u".to_string(),
2133 at: now,
2134 }),
2135 "user_signed_in",
2136 ),
2137 (
2138 Event::UserSignedUp(UserSignedUpEvent {
2139 user_id: id,
2140 username: "u".to_string(),
2141 at: now,
2142 }),
2143 "user_signed_up",
2144 ),
2145 (
2146 Event::UserSignedOut(UserSignedOutEvent {
2147 user_id: id,
2148 at: now,
2149 }),
2150 "user_signed_out",
2151 ),
2152 (
2153 Event::ProviderAccountUpdated(ProviderAccountUpdatedEvent {
2154 account_id: id,
2155 name: "perso".to_string(),
2156 change: ProviderAccountChange::TokenReplaced,
2157 at: now,
2158 }),
2159 "provider_account.updated",
2160 ),
2161 (
2162 Event::ProviderAccountUsageUpdated(ProviderAccountUsageUpdatedEvent {
2163 account_id: id,
2164 name: "perso".to_string(),
2165 windows: Vec::new(),
2166 at: now,
2167 }),
2168 "provider_account.usage_updated",
2169 ),
2170 (
2171 Event::SignalAwaited(SignalAwaitedEvent {
2172 run_id: id,
2173 step_id: id,
2174 step_name: "wait-ci".to_string(),
2175 name: "ci.done".to_string(),
2176 key: "abc".to_string(),
2177 deadline_at: now,
2178 at: now,
2179 }),
2180 "signal_awaited",
2181 ),
2182 (
2183 Event::SignalReceived(SignalReceivedEvent {
2184 signal_id: id,
2185 name: "ci.done".to_string(),
2186 key: "abc".to_string(),
2187 resumed_runs: vec![id],
2188 sender_run_id: None,
2189 sender_step_id: None,
2190 at: now,
2191 }),
2192 "signal_received",
2193 ),
2194 (
2195 Event::ScheduleOccurrencesMissed(ScheduleOccurrencesMissedEvent {
2196 schedule_id: id,
2197 workflow_name: "nightly".to_string(),
2198 reason: ScheduleMissReason::Overlap,
2199 count: 1,
2200 first: now,
2201 last: now,
2202 at: now,
2203 }),
2204 "schedule_occurrences_missed",
2205 ),
2206 (
2207 Event::WorkflowPaused(WorkflowPausedEvent {
2208 workflow_name: "w".to_string(),
2209 user_id: Some(id),
2210 at: now,
2211 }),
2212 "workflow_paused",
2213 ),
2214 (
2215 Event::WorkflowResumed(WorkflowResumedEvent {
2216 workflow_name: "w".to_string(),
2217 user_id: Some(id),
2218 at: now,
2219 }),
2220 "workflow_resumed",
2221 ),
2222 ];
2223
2224 assert_eq!(
2225 cases.len(),
2226 Event::ALL.len(),
2227 "every variant must be covered"
2228 );
2229
2230 for (event, expected_type) in cases {
2231 assert_eq!(event.event_type(), expected_type);
2232 }
2233 }
2234
2235 #[test]
2236 fn approval_escalated_serde_roundtrip() {
2237 let run_id = Uuid::now_v7();
2238 let step_id = Uuid::now_v7();
2239 let event = Event::ApprovalEscalated(ApprovalEscalatedEvent {
2240 run_id,
2241 step_id,
2242 step_name: "prod-gate".to_string(),
2243 stage: 1,
2244 policy: "escalate".to_string(),
2245 action: "reassigned to sre-oncall".to_string(),
2246 reason: "approval deadline of 3600s expired".to_string(),
2247 assignee: Some(Assignee::group("sre-oncall")),
2248 at: Utc::now(),
2249 });
2250
2251 let json = serde_json::to_string(&event).expect("serialize");
2252 assert!(
2253 json.contains("\"type\":\"approval_escalated\""),
2254 "got {json}"
2255 );
2256
2257 let back: Event = serde_json::from_str(&json).expect("deserialize");
2258 let Event::ApprovalEscalated(payload) = back else {
2259 panic!("expected an approval_escalated event");
2260 };
2261 assert_eq!(payload.run_id, run_id);
2262 assert_eq!(payload.stage, 1);
2263 assert_eq!(payload.assignee, Some(Assignee::group("sre-oncall")));
2264 }
2265
2266 #[test]
2267 fn approval_escalated_carries_run_and_step_ids() {
2268 let run_id = Uuid::now_v7();
2269 let step_id = Uuid::now_v7();
2270 let event = Event::ApprovalEscalated(ApprovalEscalatedEvent {
2271 run_id,
2272 step_id,
2273 step_name: "prod-gate".to_string(),
2274 stage: 0,
2275 policy: "notify".to_string(),
2276 action: "notified 1 target".to_string(),
2277 reason: "approval deadline of 60s expired".to_string(),
2278 assignee: None,
2279 at: Utc::now(),
2280 });
2281
2282 assert_eq!(event.run_id(), Some(run_id));
2283 assert_eq!(event.step_id(), Some(step_id));
2284 assert_eq!(event.user_id(), None);
2285 }
2286
2287 #[test]
2288 fn signal_awaited_serde_roundtrip() {
2289 let run_id = Uuid::now_v7();
2290 let step_id = Uuid::now_v7();
2291 let event = Event::SignalAwaited(SignalAwaitedEvent {
2292 run_id,
2293 step_id,
2294 step_name: "wait-ci".to_string(),
2295 name: "ci.done".to_string(),
2296 key: "abc".to_string(),
2297 deadline_at: Utc::now(),
2298 at: Utc::now(),
2299 });
2300
2301 let json = serde_json::to_string(&event).expect("serialize");
2302 assert!(json.contains("\"type\":\"signal_awaited\""), "got {json}");
2303 let back: Event = serde_json::from_str(&json).expect("deserialize");
2304 assert_eq!(back.event_type(), Event::SIGNAL_AWAITED);
2305 assert_eq!(back.run_id(), Some(run_id));
2306 assert_eq!(back.step_id(), Some(step_id));
2307 assert_eq!(back.user_id(), None);
2308 }
2309
2310 #[test]
2311 fn signal_received_serde_roundtrip() {
2312 let resumed = Uuid::now_v7();
2313 let event = Event::SignalReceived(SignalReceivedEvent {
2314 signal_id: Uuid::now_v7(),
2315 name: "ci.done".to_string(),
2316 key: "abc".to_string(),
2317 resumed_runs: vec![resumed],
2318 sender_run_id: None,
2319 sender_step_id: None,
2320 at: Utc::now(),
2321 });
2322
2323 let json = serde_json::to_string(&event).expect("serialize");
2324 assert!(json.contains("\"type\":\"signal_received\""), "got {json}");
2325 let back: Event = serde_json::from_str(&json).expect("deserialize");
2326 let Event::SignalReceived(payload) = &back else {
2327 panic!("expected a signal_received event, got {back:?}");
2328 };
2329 assert_eq!(payload.resumed_runs, vec![resumed]);
2330 assert_eq!(back.run_id(), None);
2331 assert_eq!(back.step_id(), None);
2332 assert_eq!(back.user_id(), None);
2333 assert!(!json.contains("sender_run_id"), "got {json}");
2334 }
2335
2336 #[test]
2337 fn signal_received_from_a_workflow_belongs_to_the_sender() {
2338 let (run_id, step_id) = (Uuid::now_v7(), Uuid::now_v7());
2339 let json = json!({
2340 "type": "signal_received", "signal_id": Uuid::now_v7(), "name": "ci.done",
2341 "key": "abc", "resumed_runs": [], "at": Utc::now(),
2342 "sender_run_id": run_id, "sender_step_id": step_id,
2343 });
2344 let event: Event = from_value(json).expect("deserialize");
2345 assert_eq!(event.run_id(), Some(run_id));
2346 assert_eq!(event.step_id(), Some(step_id));
2347 }
2348
2349 #[test]
2350 fn schedule_occurrences_missed_serde_roundtrip() {
2351 let schedule_id = Uuid::now_v7();
2352 let first = Utc::now();
2353 let event = Event::ScheduleOccurrencesMissed(ScheduleOccurrencesMissedEvent {
2354 schedule_id,
2355 workflow_name: "nightly".to_string(),
2356 reason: ScheduleMissReason::OutsideWindow,
2357 count: 4,
2358 first,
2359 last: first,
2360 at: first,
2361 });
2362
2363 let json = serde_json::to_string(&event).expect("serialize");
2364 assert!(
2365 json.contains("\"type\":\"schedule_occurrences_missed\""),
2366 "got {json}"
2367 );
2368 assert!(json.contains("\"reason\":\"outside_window\""), "got {json}");
2369 let back: Event = serde_json::from_str(&json).expect("deserialize");
2370 let Event::ScheduleOccurrencesMissed(payload) = &back else {
2371 panic!("expected a schedule_occurrences_missed event, got {back:?}");
2372 };
2373 assert_eq!(payload.schedule_id, schedule_id);
2374 assert_eq!(payload.reason, ScheduleMissReason::OutsideWindow);
2375 assert_eq!(payload.count, 4);
2376 assert_eq!(payload.first, first);
2377 assert!(Event::ALL.contains(&Event::SCHEDULE_OCCURRENCES_MISSED));
2378 assert_eq!(back.run_id(), None);
2379 assert_eq!(back.step_id(), None);
2380 assert_eq!(back.user_id(), None);
2381 }
2382
2383 #[test]
2384 fn legacy_approval_decisions_have_no_user_id_nor_reason() {
2385 let raw = r#"{"type":"approval_granted","run_id":"01890000-0000-7000-8000-000000000000","approved_by":"alice","at":"2026-01-01T00:00:00Z"}"#;
2386 let granted: Event = serde_json::from_str(raw).expect("deserialize");
2387 assert_eq!(granted.user_id(), None);
2388
2389 let raw = r#"{"type":"approval_rejected","run_id":"01890000-0000-7000-8000-000000000000","rejected_by":"bob","at":"2026-01-01T00:00:00Z"}"#;
2390 let rejected: Event = serde_json::from_str(raw).expect("deserialize");
2391 let Event::ApprovalRejected(rejected) = rejected else {
2392 panic!("expected approval_rejected");
2393 };
2394 assert_eq!(rejected.user_id, None);
2395 assert_eq!(rejected.reason, None);
2396 }
2397
2398 #[test]
2399 fn approval_rejected_roundtrips_the_actor_and_reason() {
2400 let user_id = Uuid::now_v7();
2401 let event = Event::ApprovalRejected(ApprovalRejectedEvent {
2402 run_id: Uuid::now_v7(),
2403 step_id: Some(Uuid::now_v7()),
2404 rejected_by: "bob".to_string(),
2405 user_id: Some(user_id),
2406 reason: Some("wrong environment".to_string()),
2407 requirement: None,
2408 at: Utc::now(),
2409 });
2410
2411 let json = serde_json::to_string(&event).expect("serialize");
2412 let back: Event = serde_json::from_str(&json).expect("deserialize");
2413 assert_eq!(back.user_id(), Some(user_id));
2414 let Event::ApprovalRejected(back) = back else {
2415 panic!("expected approval_rejected");
2416 };
2417 assert_eq!(back.reason.as_deref(), Some("wrong environment"));
2418 }
2419
2420 #[test]
2421 fn workflow_paused_and_resumed_serde_roundtrip() {
2422 let user_id = Uuid::now_v7();
2423 let now = Utc::now();
2424 let events = vec![
2425 Event::WorkflowPaused(WorkflowPausedEvent {
2426 workflow_name: "deploy".to_string(),
2427 user_id: Some(user_id),
2428 at: now,
2429 }),
2430 Event::WorkflowResumed(WorkflowResumedEvent {
2431 workflow_name: "deploy".to_string(),
2432 user_id: Some(user_id),
2433 at: now,
2434 }),
2435 ];
2436
2437 for event in events {
2438 let json = serde_json::to_string(&event).expect("serialize");
2439 assert!(
2440 json.contains(&format!("\"type\":\"{}\"", event.event_type())),
2441 "got {json}"
2442 );
2443 assert!(json.contains("\"workflow_name\":\"deploy\""), "got {json}");
2444 let back: Event = serde_json::from_str(&json).expect("deserialize");
2445 assert_eq!(back.event_type(), event.event_type());
2446 assert_eq!(back.user_id(), Some(user_id));
2447 assert_eq!(back.run_id(), None);
2448 assert_eq!(back.step_id(), None);
2449 assert!(Event::ALL.contains(&back.event_type()));
2450 }
2451 }
2452
2453 #[test]
2454 fn workflow_paused_without_user_id_deserializes() {
2455 let raw =
2456 r#"{"type":"workflow_paused","workflow_name":"deploy","at":"2026-01-01T00:00:00Z"}"#;
2457 let event: Event = serde_json::from_str(raw).expect("deserialize");
2458 assert_eq!(event.event_type(), Event::WORKFLOW_PAUSED);
2459 assert_eq!(event.user_id(), None);
2460 }
2461}