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

faucet_cli/commands/
dlq.rs
