faucet-cli 1.8.0

//! End-to-end tests for encrypted DLQ files (#207): envelopes written by the
//! jsonl sink's `encryption:` block are readable by `faucet dlq
//! inspect/replay/discard` — with an explicit key, or (for replay) picked up
//! automatically from the config's own `dlq:` block.
#![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,
    })
}

/// Write envelope lines the way a real encrypted DLQ does: through the jsonl
/// sink with an `encryption:` block.
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;

    // No key: nothing readable, but honestly reported as encrypted.
    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'");

    // With the key: fully readable.
    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));

    // Wrong key: still counted as undecryptable, never a silent skip.
    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;

    // The config declares the same encrypted jsonl DLQ it originally wrote —
    // replay derives the key from it, no --encryption-key needed.
    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(), // no explicit key — config-derived
            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);

    // The surviving line is byte-identical (still sealed) and the archive
    // holds the removed sealed line — nothing was re-encrypted or exposed.
    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");

    // Without a key, a filtered discard removes nothing (undecryptable lines
    // are preserved).
    let blind = dlq_replay::discard(
        dlq.to_str().unwrap(),
        Some("contract"),
        None,
        false,
        &DlqDecryptor::default(),
    )
    .unwrap();
    assert_eq!(blind.discarded, 0);
}