#![cfg(all(feature = "encryption", feature = "sink-jsonl"))]
use faucet_cli::auth_catalog::build_auth_catalog;
use faucet_cli::config::parse_with_extension;
use faucet_cli::dlq_replay::{self, ReplayInputs, reader::DlqDecryptor};
use faucet_core::Sink;
use serde_json::{Value, json};
use std::path::Path;
const KEY: &str = "dlq-at-rest-key";
fn envelope(reason: &str, payload: Value) -> Value {
json!({
"error": { "kind": "QualityFailure", "message": "boom" },
"reason": reason,
"payload": payload,
"ts_ms": 1_751_760_000_000i64,
"sink": "jsonl",
"pipeline": "orig",
"row": "",
"record_index": 0,
})
}
async fn write_encrypted_dlq(path: &Path, envelopes: &[Value]) {
let cfg =
faucet_sink_jsonl::JsonlSinkConfig::new(path).encryption(faucet_core::EncryptionSpec {
key: KEY.into(),
previous_keys: vec![],
algorithm: Default::default(),
});
let sink = faucet_sink_jsonl::JsonlSink::new(cfg);
sink.write_batch(envelopes).await.unwrap();
sink.flush().await.unwrap();
}
fn read_lines(path: &Path) -> Vec<Value> {
std::fs::read_to_string(path)
.unwrap_or_default()
.lines()
.filter(|l| !l.trim().is_empty())
.map(|l| serde_json::from_str(l).unwrap())
.collect()
}
#[tokio::test]
async fn inspect_reads_sealed_envelopes_with_key_and_reports_without() {
let dir = tempfile::tempdir().unwrap();
let dlq = dir.path().join("dlq.jsonl");
write_encrypted_dlq(
&dlq,
&[
envelope("quality", json!({"id": 1})),
envelope("contract", json!({"id": 2})),
],
)
.await;
let blind =
dlq_replay::inspect(dlq.to_str().unwrap(), None, 5, &DlqDecryptor::default()).unwrap();
assert_eq!(blind.total_envelopes, 0);
assert_eq!(blind.undecryptable, 2);
assert_eq!(blind.malformed, 0, "sealed lines are not 'malformed'");
let dec = DlqDecryptor::from_keys(&[KEY.to_string()]).unwrap();
let seen = dlq_replay::inspect(dlq.to_str().unwrap(), None, 5, &dec).unwrap();
assert_eq!(seen.total_envelopes, 2);
assert_eq!(seen.undecryptable, 0);
assert_eq!(seen.by_reason.get("quality"), Some(&1));
let wrong = DlqDecryptor::from_keys(&["nope".to_string()]).unwrap();
let blind = dlq_replay::inspect(dlq.to_str().unwrap(), None, 5, &wrong).unwrap();
assert_eq!(blind.undecryptable, 2);
}
#[tokio::test]
async fn replay_picks_up_the_configs_dlq_encryption_block() {
let dir = tempfile::tempdir().unwrap();
let dlq = dir.path().join("dlq.jsonl");
let out = dir.path().join("out.jsonl");
write_encrypted_dlq(&dlq, &[envelope("quality", json!({"id": 7, "ok": true}))]).await;
let cfg_yaml = format!(
concat!(
"version: 1\nname: replay\npipeline:\n",
" source: {{ type: csv, config: {{ path: /dev/null }} }}\n",
" sink: {{ type: jsonl, config: {{ path: {out} }} }}\n",
" dlq:\n sink:\n type: jsonl\n",
" config: {{ path: {dlq}, encryption: {{ key: \"{key}\" }} }}\n",
),
out = out.display(),
dlq = dlq.display(),
key = KEY,
);
let cfg = parse_with_extension(&cfg_yaml, "yaml").unwrap();
let outcome = dlq_replay::replay(
&cfg,
dlq.to_str().unwrap(),
ReplayInputs {
decryptor: Default::default(), reason: None,
failed_dlq: None,
row: None,
dry_run: false,
pipeline_name: "replay".into(),
execution: None,
auth: build_auth_catalog(None).unwrap(),
clock: chrono::Utc::now().fixed_offset(),
},
)
.await
.unwrap();
assert_eq!(outcome.candidates, 1);
assert_eq!(outcome.records_written, 1);
let rows = read_lines(&out);
assert_eq!(rows, vec![json!({"id": 7, "ok": true})]);
}
#[tokio::test]
async fn discard_filters_sealed_envelopes_and_preserves_lines_verbatim() {
let dir = tempfile::tempdir().unwrap();
let dlq = dir.path().join("dlq.jsonl");
write_encrypted_dlq(
&dlq,
&[
envelope("quality", json!({"id": 1})),
envelope("contract", json!({"id": 2})),
],
)
.await;
let before = std::fs::read_to_string(&dlq).unwrap();
let kept_line = before.lines().nth(1).unwrap().to_string();
let dec = DlqDecryptor::from_keys(&[KEY.to_string()]).unwrap();
let outcome =
dlq_replay::discard(dlq.to_str().unwrap(), Some("quality"), None, false, &dec).unwrap();
assert_eq!(outcome.discarded, 1);
let after = std::fs::read_to_string(&dlq).unwrap();
assert_eq!(after.trim(), kept_line);
let archive = std::fs::read_to_string(dir.path().join("dlq.archived.jsonl")).unwrap();
assert!(archive.contains(before.lines().next().unwrap()));
assert!(!archive.contains("\"id\":1"), "archive stays sealed");
let blind = dlq_replay::discard(
dlq.to_str().unwrap(),
Some("contract"),
None,
false,
&DlqDecryptor::default(),
)
.unwrap();
assert_eq!(blind.discarded, 0);
}