faucet_cli/commands/
notify.rs1use crate::cli::{NotifyArgs, NotifyCommand, NotifyTestArgs};
7use crate::config::PipelineConfig;
8use crate::error::{CliError, CliResult};
9use crate::notify::{Notifier, NotifyEvent};
10
11pub async fn run(args: NotifyArgs) -> CliResult<()> {
13 match args.command {
14 NotifyCommand::Test(a) => test(a).await,
15 }
16}
17
18async fn test(args: NotifyTestArgs) -> CliResult<()> {
19 let cwd = std::env::current_dir()?;
20 let env_path =
21 crate::env_loader::resolve_env_file(args.env_file.as_deref(), args.no_env_file, &cwd)?;
22 crate::env_loader::load_env_file_if_present(env_path.as_deref())?;
23
24 let path = match args.config {
25 Some(p) => p,
26 None => crate::env_loader::discover_config_path(&cwd).ok_or(CliError::NoConfigOrFromEnv)?,
27 };
28 let cfg = PipelineConfig::from_path_async(&path, None).await?;
30 if cfg.notifications.is_empty() {
31 return Err(CliError::Config(
32 "no `notifications:` block in this config — add one, or run \
33 `faucet schema notifications` to see the block's JSON Schema"
34 .to_string(),
35 ));
36 }
37
38 let notifier = Notifier::from_specs(&cfg.notifications)?
39 .expect("a non-empty notifications list yields Some(notifier)");
40 let pipeline = cfg
41 .name
42 .clone()
43 .unwrap_or_else(|| "faucet-notify-test".to_string());
44 let event = synth_event(&args.event, &pipeline)?;
45
46 println!(
47 "Firing synthetic `{}` event through {} notification rule(s)…",
48 args.event,
49 cfg.notifications.len()
50 );
51 notifier.emit(event).await;
52 println!("Done — check your channels. Any delivery failure was logged above.");
53 Ok(())
54}
55
56fn synth_event(kind: &str, pipeline: &str) -> CliResult<NotifyEvent> {
65 let run = || {
66 crate::notify::RunContext::start(
67 Some(format!("test-run-{}", uuid::Uuid::now_v7())),
68 Some(format!("test-invocation-{}", uuid::Uuid::now_v7())),
69 )
70 .finish(std::time::Instant::now())
71 };
72 Ok(match kind {
73 "run_failure" => NotifyEvent::run_failure(
74 pipeline,
75 "",
76 "test",
77 "synthetic test failure from `faucet notify test`",
78 )
79 .with_run(run()),
80 "run_success" => NotifyEvent::run_success(pipeline, "", 0).with_run(run()),
81 "sla_breach" => NotifyEvent::sla_breach(pipeline, "", "staleness", "synthetic SLA breach")
82 .with_run(run()),
83 "circuit_open" => NotifyEvent::circuit_open(pipeline, "", 5, 30).with_run(run()),
84 "contract_abort" => {
85 NotifyEvent::contract_abort(pipeline, "", "synthetic contract breach").with_run(run())
86 }
87 "dlq_threshold" => NotifyEvent::dlq_threshold(pipeline, "", 1_000_000).with_run(run()),
88 "scheduler_stuck" => NotifyEvent::scheduler_stuck(pipeline, "synthetic scheduler-stuck"),
89 other => {
90 return Err(CliError::Config(format!(
91 "unknown --event `{other}` (expected one of: run_failure, run_success, \
92 sla_breach, circuit_open, contract_abort, dlq_threshold, scheduler_stuck)"
93 )));
94 }
95 })
96}
97
98#[cfg(test)]
99mod tests {
100 use super::*;
101
102 #[test]
103 fn synth_event_maps_known_kinds() {
104 use crate::notify::EventKind;
105 assert_eq!(
106 synth_event("run_failure", "p").unwrap().kind,
107 EventKind::RunFailure
108 );
109 assert_eq!(
110 synth_event("scheduler_stuck", "p").unwrap().kind,
111 EventKind::SchedulerStuck
112 );
113 assert_eq!(
115 synth_event("dlq_threshold", "p")
116 .unwrap()
117 .details
118 .get("records_dlq")
119 .and_then(|v| v.as_u64()),
120 Some(1_000_000)
121 );
122 }
123
124 #[test]
125 fn synth_event_rejects_unknown_kind() {
126 assert!(synth_event("nope", "p").is_err());
127 }
128
129 #[test]
130 fn synth_events_carry_run_identity_except_scheduler_stuck() {
131 for kind in [
134 "run_failure",
135 "run_success",
136 "sla_breach",
137 "circuit_open",
138 "contract_abort",
139 "dlq_threshold",
140 ] {
141 let e = synth_event(kind, "p").unwrap();
142 let run = e
143 .run
144 .as_ref()
145 .unwrap_or_else(|| panic!("{kind} needs a run context"));
146 assert!(run.run_id.is_some(), "{kind} run_id");
147 assert!(run.invocation_id.is_some(), "{kind} invocation_id");
148 assert_ne!(run.run_id, run.invocation_id, "{kind} ids must differ");
149 assert!(run.finished_at.is_some(), "{kind} finished_at");
150 assert!(run.duration.is_some(), "{kind} duration");
151 }
152 assert!(synth_event("scheduler_stuck", "p").unwrap().run.is_none());
154 }
155}