faucet_cli/commands/
dlq.rs

1//! `faucet dlq` — inspect / replay / discard dead-letter-queue envelopes.
2//!
3//! Thin command layer over [`crate::dlq_replay`]: it loads config (for
4//! `replay`), calls the orchestration functions, and renders the returned
5//! summaries for the terminal (or as JSON with `--json`).
6
7use crate::cli::{DlqArgs, DlqCommand, DlqDiscardArgs, DlqInspectArgs, DlqReplayArgs};
8use crate::config::PipelineConfig;
9use crate::dlq_replay::{self, ReplayInputs};
10use crate::error::{CliError, CliResult};
11use chrono::{DateTime, Utc};
12
13/// Serialize a result struct to pretty JSON, mapping the (practically
14/// impossible) serialization error to a typed CLI error.
15fn to_json<T: serde::Serialize>(value: &T) -> CliResult<String> {
16    serde_json::to_string_pretty(value)
17        .map_err(|e| CliError::Internal(format!("serializing JSON output: {e}")))
18}
19
20/// Dispatch a `faucet dlq` subcommand.
21pub async fn run(args: DlqArgs) -> CliResult<()> {
22    match args.command {
23        DlqCommand::Inspect(a) => inspect(a),
24        DlqCommand::Replay(a) => replay(a).await,
25        DlqCommand::Discard(a) => discard(a),
26    }
27}
28
29fn inspect(args: DlqInspectArgs) -> CliResult<()> {
30    let summary = dlq_replay::inspect(&args.location, args.reason.as_deref(), args.limit)?;
31    if args.json {
32        println!("{}", to_json(&summary)?);
33        return Ok(());
34    }
35    println!("DLQ inspect: {}", summary.location);
36    println!(
37        "  files read: {}   envelopes: {}   malformed: {}   non-envelope: {}",
38        summary.files_read, summary.total_envelopes, summary.malformed, summary.non_envelope
39    );
40    if !summary.by_reason.is_empty() {
41        println!("  by reason:");
42        for (reason, count) in &summary.by_reason {
43            println!("    {reason:<14} {count}");
44        }
45    }
46    if !summary.by_error_kind.is_empty() {
47        println!("  by error kind:");
48        for (kind, count) in &summary.by_error_kind {
49            println!("    {kind:<20} {count}");
50        }
51    }
52    if !summary.sample.is_empty() {
53        println!(
54            "  sample ({} of {}):",
55            summary.sample.len(),
56            summary.total_envelopes
57        );
58        for env in &summary.sample {
59            let reason = env.reason.as_deref().unwrap_or("?");
60            let kind = env.error_kind.as_deref().unwrap_or("?");
61            let msg = env.error_message.as_deref().unwrap_or("");
62            println!("    [{reason}/{kind}] {msg}");
63            println!("      {}", env.payload);
64        }
65    }
66    Ok(())
67}
68
69async fn replay(args: DlqReplayArgs) -> CliResult<()> {
70    let cwd = std::env::current_dir()?;
71    let env_path =
72        crate::env_loader::resolve_env_file(args.env_file.as_deref(), args.no_env_file, &cwd)?;
73    crate::env_loader::load_env_file_if_present(env_path.as_deref())?;
74    let path = match args.config {
75        Some(p) => p,
76        None => crate::env_loader::discover_config_path(&cwd).ok_or(CliError::NoConfigOrFromEnv)?,
77    };
78
79    let cfg = PipelineConfig::from_path_async(&path, args.profile.as_deref()).await?;
80    crate::obs::install(&cfg)?;
81
82    let pipeline_name = cfg.name.clone().unwrap_or_else(|| {
83        path.file_stem()
84            .and_then(|s| s.to_str())
85            .unwrap_or("pipeline")
86            .to_owned()
87    });
88    let auth = crate::auth_catalog::build_auth_catalog(cfg.auth.as_ref())?;
89
90    let outcome = dlq_replay::replay(
91        &cfg,
92        &args.from,
93        ReplayInputs {
94            reason: args.reason.as_deref(),
95            failed_dlq: args.failed_dlq.as_deref(),
96            row: args.row.as_deref(),
97            dry_run: args.dry_run,
98            pipeline_name,
99            execution: cfg.execution.clone(),
100            auth,
101            clock: Utc::now().fixed_offset(),
102        },
103    )
104    .await?;
105
106    faucet_core::shutdown_otel();
107
108    if args.json {
109        println!("{}", to_json(&outcome)?);
110        return Ok(());
111    }
112    if outcome.dry_run {
113        println!(
114            "DLQ replay (dry-run): {} candidate record(s) from {} would be re-fed; \
115             {} would reach the sink. Failures would go to {}.",
116            outcome.candidates, args.from, outcome.records_written, outcome.failed_dlq
117        );
118    } else {
119        println!(
120            "DLQ replay: {} candidate record(s) from {} re-fed; {} written to the sink. \
121             Rows that failed again went to {}.",
122            outcome.candidates, args.from, outcome.records_written, outcome.failed_dlq
123        );
124    }
125    Ok(())
126}
127
128fn discard(args: DlqDiscardArgs) -> CliResult<()> {
129    let before_ms = match args.before.as_deref() {
130        Some(s) => Some(parse_before(s, Utc::now())?),
131        None => None,
132    };
133    let outcome = dlq_replay::discard(
134        &args.location,
135        args.reason.as_deref(),
136        before_ms,
137        args.delete,
138    )?;
139    if args.json {
140        println!("{}", to_json(&outcome)?);
141        return Ok(());
142    }
143    if args.delete {
144        println!(
145            "DLQ discard: deleted {} envelope(s) across {} file(s).",
146            outcome.discarded, outcome.files_rewritten
147        );
148    } else {
149        println!(
150            "DLQ discard: archived {} envelope(s) across {} file(s){}.",
151            outcome.discarded,
152            outcome.files_rewritten,
153            if outcome.archived_to.is_empty() {
154                String::new()
155            } else {
156                format!(" → {}", outcome.archived_to.join(", "))
157            }
158        );
159    }
160    Ok(())
161}
162
163/// Parse a `--before` value into an epoch-millis cutoff: an RFC3339 timestamp
164/// (`2026-06-01T00:00:00Z`), or a relative age (`7d` / `24h` / `30m` / `45s`)
165/// subtracted from `now`.
166fn parse_before(s: &str, now: DateTime<Utc>) -> CliResult<i64> {
167    if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
168        return Ok(dt.timestamp_millis());
169    }
170    if let Some(cutoff) = parse_relative_age(s, now) {
171        return Ok(cutoff);
172    }
173    Err(CliError::Config(format!(
174        "--before '{s}' is not an RFC3339 timestamp or a relative age like 7d / 24h / 30m / 45s"
175    )))
176}
177
178/// Parse a relative age like `7d` into an epoch-millis cutoff (`now - age`).
179/// Returns `None` if `s` is not `<positive-int><d|h|m|s>`.
180fn parse_relative_age(s: &str, now: DateTime<Utc>) -> Option<i64> {
181    let (num, unit) = s.split_at(s.len().checked_sub(1)?);
182    let n: i64 = num.parse().ok()?;
183    if n <= 0 {
184        return None;
185    }
186    let secs = match unit {
187        "d" => n.checked_mul(86_400)?,
188        "h" => n.checked_mul(3_600)?,
189        "m" => n.checked_mul(60)?,
190        "s" => n,
191        _ => return None,
192    };
193    Some((now - chrono::Duration::seconds(secs)).timestamp_millis())
194}
195
196#[cfg(test)]
197mod tests {
198    use super::*;
199
200    fn now() -> DateTime<Utc> {
201        DateTime::parse_from_rfc3339("2026-07-06T00:00:00Z")
202            .unwrap()
203            .with_timezone(&Utc)
204    }
205
206    #[test]
207    fn parse_before_rfc3339() {
208        let ms = parse_before("2026-06-01T00:00:00Z", now()).unwrap();
209        assert_eq!(
210            ms,
211            DateTime::parse_from_rfc3339("2026-06-01T00:00:00Z")
212                .unwrap()
213                .timestamp_millis()
214        );
215    }
216
217    #[test]
218    fn parse_before_relative_ages() {
219        let now = now();
220        assert_eq!(
221            parse_before("1d", now).unwrap(),
222            (now - chrono::Duration::seconds(86_400)).timestamp_millis()
223        );
224        assert_eq!(
225            parse_before("2h", now).unwrap(),
226            (now - chrono::Duration::seconds(7_200)).timestamp_millis()
227        );
228        assert_eq!(
229            parse_before("30m", now).unwrap(),
230            (now - chrono::Duration::seconds(1_800)).timestamp_millis()
231        );
232    }
233
234    #[test]
235    fn parse_before_rejects_garbage() {
236        assert!(parse_before("soon", now()).is_err());
237        assert!(parse_before("0d", now()).is_err());
238        assert!(parse_before("-5d", now()).is_err());
239        assert!(parse_before("7y", now()).is_err());
240    }
241}
faucet_cli/commands/dlq.rs

faucet_cli/commands/
dlq.rs
