//! What `Writer::open` does with the log it finds: refuse a broken chain, recover a torn tail,
//! accept a break when told to. Do not edit. The fixture logs are in
//! `crates/proto/tests/fixtures/audit/`; each test works on a copy.

#[path = "support/audit_dir.rs"]
mod audit_dir;

use audit_dir::{D1, D2, TempDir, denied, lines, snapshot};
use brokerd::audit::{AuditError, Writer, verify_dir};
use proto::{AuditEvent, AuditRecord, Location, Timestamp};

type Case = (
    &'static str,
    Option<&'static [&'static str]>,
    &'static str,
    u64,
    &'static str,
);

fn at(file: &str, line: u64) -> Location {
    Location {
        file: file.to_string(),
        line,
    }
}

/// An ordinary start checks the latest file only, so each damaged file is copied alone: it is
/// then the latest. Nothing may be written to a log that does not verify.
#[test]
fn a_broken_chain_refuses_to_start_and_writes_nothing() {
    let parse = "does not parse as an audit record";
    // (case, the files to copy, then the failure's file, line and text)
    let cases: [Case; 9] = [
        (
            "changed-byte",
            Some(&[D1]),
            D1,
            4,
            "prev is not the hash of the line before",
        ),
        ("deleted-line", Some(&[D1]), D1, 3, "seq is 3, expected 2"),
        ("swapped-lines", Some(&[D1]), D1, 2, "seq is 2, expected 1"),
        ("seq-gap", None, D1, 3, "seq is 3, expected 2"),
        ("cut-short", Some(&[D1]), D1, 3, parse),
        // Both files: the latest does not chain from the last line of the one before.
        (
            "file-not-chained",
            None,
            D2,
            1,
            "does not chain from the last line of the file before",
        ),
        (
            "break-without-failure",
            None,
            D2,
            6,
            "an accepted break with no failure before it",
        ),
        ("recovery-wrong-hash", None, D2, 6, parse),
        ("torn-recovery", None, D2, 6, parse),
    ];
    for (case, only, file, line, what) in cases {
        let dir = TempDir::case(case, only);
        let before = snapshot(&dir.path);
        let error = Writer::open(&dir.path, false)
            .err()
            .unwrap_or_else(|| panic!("{case}: started"));
        let AuditError::Broken(failure) = &error else {
            panic!("{case}: {error}");
        };
        assert_eq!(
            (failure.file.as_str(), failure.line, failure.what.as_str()),
            (file, line, what),
            "{case}"
        );
        let text = error.to_string();
        assert!(
            text.starts_with(&format!("{file}:{line}: {what}")),
            "{case}: {text}"
        );
        assert!(
            text.ends_with("see docs/runbook.md#audit-chain-broken"),
            "{case}: {text}"
        );
        assert_eq!(
            snapshot(&dir.path),
            before,
            "{case}: the log was written to"
        );
    }
}

/// A torn tail is recovered: the torn bytes stay, a newline ends them if one is missing, and a
/// `Recovery` record follows in the same file, whatever today's date is.
#[test]
fn a_torn_tail_is_recovered() {
    // (case, the torn line's file and number, newline already there)
    let cases = [
        ("torn-tail", D2, 6, false),
        ("torn-tail-complete-json", D2, 6, false),
        ("torn-unparseable-newline", D2, 6, true),
        ("torn-first-line", D2, 1, false),
    ];
    for (case, file, line, has_newline) in cases {
        let dir = TempDir::case(case, None);
        let before = snapshot(&dir.path);
        let opened = Writer::open(&dir.path, false).unwrap_or_else(|e| panic!("{case}: {e}"));
        assert!(opened.recovered, "{case}");
        assert!(opened.accepted.is_none(), "{case}");

        let after = snapshot(&dir.path);
        assert_eq!(
            after.len(),
            before.len(),
            "{case}: the Recovery went into a new file"
        );
        let (old, new) = (&before[file], &after[file]);
        assert!(
            new.starts_with(old),
            "{case}: bytes already on disk were changed"
        );
        let added = &new[old.len()..];
        // One newline to end the torn line if it had none, then one line.
        let added = if has_newline {
            added
        } else {
            added.strip_prefix(b"\n").expect(case)
        };
        assert_eq!(added.iter().filter(|b| **b == b'\n').count(), 1, "{case}");
        let record: AuditRecord =
            serde_json::from_slice(added.strip_suffix(b"\n").expect(case)).expect(case);
        assert!(
            matches!(record.event, AuditEvent::Recovery { .. }),
            "{case}"
        );

        let report = verify_dir(&dir.path).unwrap();
        assert_eq!(report.failure, None, "{case}");
        assert_eq!(report.torn_tail, None, "{case}");
        assert_eq!(report.recoveries, vec![at(file, line)], "{case}");

        // The chain goes on from the Recovery, and the next start finds nothing to recover.
        let mut writer = opened.writer;
        assert_eq!(
            writer.append(Timestamp::now(), denied(9)).unwrap(),
            record.seq + 1,
            "{case}"
        );
        drop(writer);
        let opened = Writer::open(&dir.path, false).unwrap();
        assert!(!opened.recovered, "{case}");
        assert_eq!(verify_dir(&dir.path).unwrap().failure, None, "{case}");
    }
}

/// Damage in an older file is not seen by an ordinary start. `bxctl audit verify` sees it, and
/// `--accept-break` must too: it verifies the whole log.
#[test]
fn a_break_in_an_older_file_can_be_accepted() {
    let dir = TempDir::case("changed-byte", None);
    let before = snapshot(&dir.path);
    drop(Writer::open(&dir.path, false).expect("the latest file verifies"));
    assert_eq!(snapshot(&dir.path), before);
    let failure = verify_dir(&dir.path).unwrap().failure.unwrap();
    assert_eq!((failure.file.as_str(), failure.line), (D1, 4));

    let opened = Writer::open(&dir.path, true).unwrap();
    assert!(!opened.recovered);
    assert_eq!(
        opened.accepted.as_ref().map(|f| (f.file.as_str(), f.line)),
        Some((D1, 4))
    );
    let after = snapshot(&dir.path);
    assert_eq!(after[D1], before[D1], "nothing is repaired");
    assert!(after[D2].starts_with(&before[D2]));
    let last: AuditRecord = serde_json::from_str(lines(&dir.path, D2).last().unwrap()).unwrap();
    assert_eq!(
        last.event,
        AuditEvent::AcceptedBreak {
            file: D1.to_string(),
            line: 4,
            last_good: failure.last_good,
        }
    );
    assert_eq!(
        last.seq, 10,
        "seq 3 for the failing line, and seven lines to the break"
    );

    let report = verify_dir(&dir.path).unwrap();
    assert_eq!(report.failure, None);
    assert_eq!(report.accepted_breaks, vec![at(D2, 6)]);

    let mut writer = opened.writer;
    assert_eq!(writer.append(Timestamp::now(), denied(9)).unwrap(), 11);
    drop(writer);

    // The next ordinary start resumes at the latest file and meets a break that names a file
    // it has not read.
    let mut writer = Writer::open(&dir.path, false).unwrap().writer;
    assert_eq!(writer.append(Timestamp::now(), denied(10)).unwrap(), 12);
    assert_eq!(verify_dir(&dir.path).unwrap().failure, None);
}

#[test]
fn a_break_in_the_latest_file_can_be_accepted() {
    // (case, failing line, seq of the break record)
    for (case, line, seq) in [("recovery-wrong-hash", 6, 12), ("torn-recovery", 6, 12)] {
        let dir = TempDir::case(case, None);
        let before = snapshot(&dir.path);
        let opened = Writer::open(&dir.path, true).unwrap_or_else(|e| panic!("{case}: {e}"));
        assert_eq!(
            opened.accepted.as_ref().map(|f| f.line),
            Some(line),
            "{case}"
        );
        let after = snapshot(&dir.path);
        assert!(after[D2].starts_with(&before[D2]), "{case}");

        // torn-recovery ends without a newline: the break record must start on its own line.
        let all = lines(&dir.path, D2);
        let last: AuditRecord = serde_json::from_str(all.last().unwrap()).expect(case);
        assert!(
            matches!(last.event, AuditEvent::AcceptedBreak { .. }),
            "{case}"
        );
        assert_eq!(last.seq, seq, "{case}");
        assert_eq!(all.len(), 8, "{case}");

        let report = verify_dir(&dir.path).unwrap();
        assert_eq!(report.failure, None, "{case}");
        assert_eq!(report.accepted_breaks, vec![at(D2, 8)], "{case}");
        drop(opened);
        assert!(Writer::open(&dir.path, false).is_ok(), "{case}");
    }
}

/// The short check is a shortcut and never the last word: when it cannot be made, or fails, the
/// whole log is verified and that verdict stands. Here the last line of the older file is the
/// damage, so there is nothing to resume from.
#[test]
fn an_accepted_break_at_the_end_of_an_older_file_does_not_stop_later_starts() {
    let dir = TempDir::case("good", None);
    let day1 = std::fs::read_to_string(dir.path.join(D1)).unwrap();
    let cut = format!("{}\n", &day1[..day1.len() - 40]);
    std::fs::write(dir.path.join(D1), cut).unwrap();

    let error = Writer::open(&dir.path, false).unwrap_err();
    let AuditError::Broken(failure) = &error else {
        panic!("{error}");
    };
    assert_eq!((failure.file.as_str(), failure.line), (D1, 5));

    drop(Writer::open(&dir.path, true).unwrap());
    let mut writer = Writer::open(&dir.path, false)
        .expect("the break was accepted")
        .writer;
    writer.append(Timestamp::now(), denied(9)).unwrap();
    assert_eq!(verify_dir(&dir.path).unwrap().failure, None);
}

#[test]
fn accept_break_with_nothing_to_accept_is_an_error() {
    for case in ["good", "torn-tail"] {
        let dir = TempDir::case(case, None);
        let before = snapshot(&dir.path);
        let error = Writer::open(&dir.path, true).unwrap_err();
        assert!(
            matches!(error, AuditError::NothingToAccept),
            "{case}: {error}"
        );
        assert!(error.to_string().starts_with("nothing to accept"), "{case}");
        assert_eq!(
            snapshot(&dir.path),
            before,
            "{case}: the log was written to"
        );
    }
}

/// A second failure after an accepted break needs its own break.
#[test]
fn damage_after_a_break_is_a_new_failure() {
    let dir = TempDir::case("accepted-break", None);
    drop(Writer::open(&dir.path, false).expect("the fixture verifies"));
    let mut text = std::fs::read_to_string(dir.path.join(D1)).unwrap();
    text.push_str("{}\n{}\n");
    std::fs::write(dir.path.join(D1), text).unwrap();

    let error = Writer::open(&dir.path, false).unwrap_err();
    let AuditError::Broken(failure) = &error else {
        panic!("{error}");
    };
    assert_eq!((failure.file.as_str(), failure.line), (D1, 8));
    drop(Writer::open(&dir.path, true).unwrap());
    let report = verify_dir(&dir.path).unwrap();
    assert_eq!(report.failure, None);
    assert_eq!(report.accepted_breaks, vec![at(D1, 6), at(D1, 10)]);
}