powercut 0.1.1

//! The runner: workload → ack → capture → reboot → check.

use std::io;
use std::path::{Path, PathBuf};

use crate::capture::{CrashState, Tier};
use crate::report::{Finding, Report, Violation, ViolationKind};
use crate::witness::{Coverage, DurabilityWitness};
use crate::Result;

/// One measurement: run a workload to an ack point, capture what survives at a
/// [`Tier`], and hand the captured directory to your checker.
///
/// The four things you supply are deliberately small, because a durability
/// harness that needs an ORM excludes the engines that most need testing:
///
/// 1. **the store directory** — every file your engine keeps state in;
/// 2. **a witness** — [`SyncJournal`](crate::SyncJournal), usually;
/// 3. **a workload closure** — whose return value *is* the ack;
/// 4. **a checker closure** — reopen from the captured copy and read back.
///
/// ```
/// use powercut::{Checker, DurabilityWitness, ScratchDir, Scenario, SyncJournal, Tier, Verdict};
///
/// let store = ScratchDir::new("doc-scenario").unwrap();
/// let journal = SyncJournal::complete();
/// let db = store.path().join("app.log");
///
/// let report = Scenario::new(store.path(), journal.clone())
///     .label("append-then-fsync")
///     .tier(Tier::PowerCut)
///     .ack(|| {
///         std::fs::write(&db, b"balance=100").unwrap();
///         journal.sync_path(&db).unwrap();      // the durable-ack
///         journal.sync_dir(store.path()).unwrap();
///         "balance=100"
///     })
///     .unwrap()
///     .check(|crash, acked| {
///         let mut c = Checker::new();
///         let back = std::fs::read_to_string(crash.dir().join("app.log")).ok();
///         c.survives("app.log", Some(acked.to_string()), back);
///         c.finish()
///     });
///
/// assert_eq!(report.verdict(), Verdict::Survived);
/// ```
#[derive(Debug)]
pub struct Scenario<W> {
    store: PathBuf,
    witness: W,
    tier: Tier,
    label: String,
}

impl<W: DurabilityWitness> Scenario<W> {
    /// Point a scenario at a store directory and a witness.
    ///
    /// The tier defaults to [`Tier::PowerCut`], because the weaker tier is the
    /// one people measure by accident.
    ///
    /// ```
    /// # use powercut::{ScratchDir, Scenario, SyncJournal, Tier};
    /// let store = ScratchDir::new("doc-new").unwrap();
    /// let s = Scenario::new(store.path(), SyncJournal::complete());
    /// assert_eq!(s.selected_tier(), Tier::PowerCut);
    /// ```
    pub fn new(store: impl AsRef<Path>, witness: W) -> Self {
        let store = store.as_ref().to_path_buf();
        let label = store
            .file_name()
            .map(|n| n.to_string_lossy().into_owned())
            .unwrap_or_else(|| "unnamed".to_string());
        Self {
            store,
            witness,
            tier: Tier::PowerCut,
            label,
        }
    }

    /// Choose the failure being modelled.
    ///
    /// ```
    /// # use powercut::{ScratchDir, Scenario, SyncJournal, Tier};
    /// # let store = ScratchDir::new("doc-tier").unwrap();
    /// let s = Scenario::new(store.path(), SyncJournal::complete()).tier(Tier::ProcessCrash);
    /// assert_eq!(s.selected_tier(), Tier::ProcessCrash);
    /// ```
    #[must_use]
    pub fn tier(mut self, tier: Tier) -> Self {
        self.tier = tier;
        self
    }

    /// Name this case. Baselines are keyed on it, so keep it stable.
    ///
    /// ```
    /// # use powercut::{ScratchDir, Scenario, SyncJournal};
    /// # let store = ScratchDir::new("doc-label").unwrap();
    /// let s = Scenario::new(store.path(), SyncJournal::complete()).label("sqlite/synchronous=OFF");
    /// assert_eq!(s.selected_label(), "sqlite/synchronous=OFF");
    /// ```
    #[must_use]
    pub fn label(mut self, label: impl Into<String>) -> Self {
        self.label = label.into();
        self
    }

    /// The tier this scenario will measure.
    ///
    /// ```
    /// # use powercut::{ScratchDir, Scenario, SyncJournal, Tier};
    /// # let store = ScratchDir::new("doc-selected-tier").unwrap();
    /// # let s = Scenario::new(store.path(), SyncJournal::complete());
    /// assert_eq!(s.selected_tier(), Tier::PowerCut);
    /// ```
    #[must_use]
    pub fn selected_tier(&self) -> Tier {
        self.tier
    }

    /// The label this scenario will report under.
    ///
    /// ```
    /// # use powercut::{ScratchDir, Scenario, SyncJournal};
    /// # let store = ScratchDir::new("doc-selected-label").unwrap();
    /// # let s = Scenario::new(store.path(), SyncJournal::complete()).label("case-1");
    /// assert_eq!(s.selected_label(), "case-1");
    /// ```
    #[must_use]
    pub fn selected_label(&self) -> &str {
        &self.label
    }

    /// Run `workload`, then cut the power the instant it returns.
    ///
    /// Whatever `workload` returns **is the ack** — the promise the surviving
    /// bytes have to honour. Return the id you handed the caller, the balance
    /// you displayed, the row you said you wrote.
    ///
    /// At [`Tier::PowerCut`] a tier-1 byte copy is captured at the same
    /// instant as a reference, so the report can tell *"never reached the
    /// filesystem"* apart from *"reached the filesystem but was never fsynced"*.
    ///
    /// # Errors
    ///
    /// Fails if the store directory does not exist or the capture cannot be
    /// written.
    ///
    /// ```
    /// # use powercut::{ScratchDir, Scenario, SyncJournal};
    /// # let store = ScratchDir::new("doc-ack").unwrap();
    /// let acked = Scenario::new(store.path(), SyncJournal::complete())
    ///     .ack(|| 42_i64)
    ///     .unwrap();
    /// assert_eq!(*acked.ack(), 42);
    /// ```
    pub fn ack<T>(self, workload: impl FnOnce() -> T) -> Result<Acked<T, W>> {
        let ack = workload();
        let measured = CrashState::capture(&self.store, self.tier, &self.witness)?;
        let reference = if self.tier == Tier::PowerCut {
            Some(CrashState::capture(
                &self.store,
                Tier::ProcessCrash,
                &self.witness,
            )?)
        } else {
            None
        };
        Ok(Acked {
            store: self.store,
            witness: self.witness,
            tier: self.tier,
            label: self.label,
            ack,
            measured,
            reference,
            control: None,
        })
    }
}

/// A workload that has acked, with the crash state already captured.
///
/// Nothing happens to the live store between the ack and the capture: no
/// graceful shutdown, no checkpoint, no flush. That is the whole point.
///
/// [`Scenario::ack`] hands you one of these rather than a [`Report`] so that
/// the two things that happen after the capture — adding a teeth control, and
/// running the checker — stay separate steps you can inspect between.
///
/// ```
/// use powercut::{Checker, ScratchDir, Scenario, SyncJournal, Tier, Verdict};
///
/// let store = ScratchDir::new("doc-acked").unwrap();
/// let journal = SyncJournal::complete();
/// let db = store.path().join("orders.db");
///
/// let acked = Scenario::new(store.path(), journal.clone())
///     .label("orders/durable-ack")
///     .tier(Tier::PowerCut)
///     .ack(|| {
///         std::fs::write(&db, b"order=7").unwrap();
///         journal.sync_path(&db).unwrap();
///         7_i64                              // the id you handed the caller
///     })
///     .unwrap();
///
/// // The power is already out. Nothing below can change what survived.
/// assert_eq!(*acked.ack(), 7);
/// assert_eq!(acked.crash_state().manifest().len(), 1);
///
/// let report = acked.check(|crash, id| {
///     let mut c = Checker::new();
///     let back = std::fs::read(crash.dir().join("orders.db")).ok();
///     c.survives("order", Some(format!("order={id}").into_bytes()), back);
///     c.finish()
/// });
/// assert_eq!(report.verdict(), Verdict::Survived);
/// ```
#[derive(Debug)]
pub struct Acked<T, W> {
    store: PathBuf,
    witness: W,
    tier: Tier,
    label: String,
    ack: T,
    measured: CrashState,
    reference: Option<CrashState>,
    control: Option<CrashState>,
}

impl<T, W: DurabilityWitness> Acked<T, W> {
    /// What the workload acked.
    ///
    /// This is the promise the surviving bytes have to honour, so it is
    /// whatever your workload returned: the id you handed the caller, the
    /// balance you displayed, the row you said you wrote.
    ///
    /// ```
    /// use powercut::{ScratchDir, Scenario, SyncJournal};
    ///
    /// let store = ScratchDir::new("doc-acked-ack").unwrap();
    /// let journal = SyncJournal::complete();
    /// let ledger = store.path().join("ledger");
    ///
    /// let acked = Scenario::new(store.path(), journal.clone())
    ///     .ack(|| {
    ///         std::fs::write(&ledger, b"acct-1 balance=100\n").unwrap();
    ///         journal.sync_path(&ledger).unwrap();
    ///         ("acct-1".to_string(), 100_u64)     // what the API told the caller
    ///     })
    ///     .unwrap();
    ///
    /// let (account, balance) = acked.ack();
    /// assert_eq!(account, "acct-1");
    /// assert_eq!(*balance, 100);
    /// ```
    #[must_use]
    pub fn ack(&self) -> &T {
        &self.ack
    }

    /// The crash state that will be checked.
    ///
    /// Reach for this to assert on the durable set *itself* — which files
    /// carried a warrant and which quietly did not — rather than on what a
    /// reopened store can read back.
    ///
    /// ```
    /// use powercut::{ScratchDir, Scenario, SyncJournal, Warrant};
    ///
    /// let store = ScratchDir::new("doc-acked-crash-state").unwrap();
    /// let journal = SyncJournal::complete();
    ///
    /// let acked = Scenario::new(store.path(), journal.clone())
    ///     .ack(|| {
    ///         std::fs::write(store.path().join("data"), b"acked").unwrap();
    ///         std::fs::write(store.path().join("index"), b"derived").unwrap();
    ///         journal.sync_path(store.path().join("data")).unwrap();   // only one of them
    ///     })
    ///     .unwrap();
    ///
    /// let crash = acked.crash_state();
    /// let survivors: Vec<_> = crash.manifest().iter().map(|f| f.path.clone()).collect();
    /// assert_eq!(survivors, [std::path::PathBuf::from("data")]);
    /// assert!(matches!(crash.manifest()[0].warrant, Warrant::Synced(_)));
    /// assert!(!crash.dir().join("index").exists(), "never fsynced, never durable");
    /// ```
    #[must_use]
    pub fn crash_state(&self) -> &CrashState {
        &self.measured
    }

    /// **The anti-vacuity control, and it is not decoration.**
    ///
    /// A durability test's dangerous failure is not "red when it should be
    /// green". It is **red for a reason that is not durability** — a store
    /// path that misses a sidecar, a copy that will not reopen, a checker
    /// looking in the wrong place. All three look exactly like data loss.
    ///
    /// `flush` must make the store's state durable by whatever route your
    /// system offers, recording those syncs into the same witness (usually
    /// [`SyncJournal::sync_path`](crate::SyncJournal::sync_path) over every
    /// store file). `powercut` then captures a second crash state at the same
    /// tier and runs the same checker against it. That run **must** pass. If
    /// it does not, the verdict is [`Verdict::HarnessBroken`](crate::Verdict::HarnessBroken)
    /// and never [`Verdict::Lost`](crate::Verdict::Lost) — because a broken
    /// harness must not be able to accuse a database.
    ///
    /// # Errors
    ///
    /// Fails if `flush` fails or the control capture cannot be written.
    ///
    /// ```
    /// use powercut::{Checker, DurabilityWitness, ScratchDir, Scenario, SyncJournal, Verdict};
    ///
    /// let store = ScratchDir::new("doc-teeth").unwrap();
    /// let journal = SyncJournal::complete();
    /// let db = store.path().join("app.log");
    ///
    /// let report = Scenario::new(store.path(), journal.clone())
    ///     .label("no-fsync")
    ///     .ack(|| std::fs::write(&db, b"acked").unwrap())
    ///     .unwrap()
    ///     .with_teeth_control(|| { journal.sync_path(&db)?; Ok(()) })
    ///     .unwrap()
    ///     .check(|crash, ()| {
    ///         let mut c = Checker::new();
    ///         c.survives("app.log", Some(b"acked".to_vec()), std::fs::read(crash.dir().join("app.log")).ok());
    ///         c.finish()
    ///     });
    ///
    /// // Red, and the control proves the red is about durability.
    /// assert_eq!(report.verdict(), Verdict::Lost);
    /// ```
    pub fn with_teeth_control(mut self, flush: impl FnOnce() -> io::Result<()>) -> Result<Self> {
        flush().map_err(|e| crate::Error::io(Some(self.store.clone()), e))?;
        self.control = Some(CrashState::capture(&self.store, self.tier, &self.witness)?);
        Ok(self)
    }

    /// Reboot from the captured state and check the ack's promise.
    ///
    /// `check` gets the [`CrashState`] and the ack value, and returns every
    /// violation it found. An empty vector means the contract held. Build the
    /// vector with [`Checker`](crate::Checker) if you want the baseline-first
    /// discipline for free.
    ///
    /// ```
    /// # use powercut::{ScratchDir, Scenario, SyncJournal, Verdict};
    /// # let store = ScratchDir::new("doc-check").unwrap();
    /// let report = Scenario::new(store.path(), SyncJournal::complete())
    ///     .ack(|| ())
    ///     .unwrap()
    ///     .check(|_crash, ()| Vec::new());
    /// assert_eq!(report.verdict(), Verdict::Survived);
    /// ```
    #[must_use]
    pub fn check(self, check: impl Fn(&CrashState, &T) -> Vec<Violation>) -> Report {
        let violations = check(&self.measured, &self.ack);
        let reference_violations = self.reference.as_ref().map(|r| check(r, &self.ack));
        let control_violations = self.control.as_ref().map(|c| check(c, &self.ack));
        let coverage = self.witness.coverage();

        let finding = decide(
            &violations,
            control_violations.as_deref(),
            &coverage,
            self.tier,
        );

        Report {
            label: self.label,
            tier: self.tier,
            finding,
            coverage,
            manifest: self.measured.manifest().to_vec(),
            manifest_text: self.measured.manifest_text(),
            notes: self.measured.notes().to_vec(),
            reference_violations,
            control_violations,
            used_assumption: self.measured.used_assumption(),
        }
    }
}

/// The adjudication rules, in one place so they can be argued with.
fn decide(
    violations: &[Violation],
    control: Option<&[Violation]>,
    coverage: &Coverage,
    tier: Tier,
) -> Finding {
    // 1. A control that fails outranks everything: the harness is the suspect.
    if let Some(control) = control {
        if !control.is_empty() {
            return Finding::HarnessBroken {
                why: "the teeth control failed: with the store explicitly flushed, the same \
                      capture and the same checker still could not find the acked data, so \
                      this run says nothing about durability. In order of how often it is the \
                      cause: (1) the flush closure does not actually force the engine's own \
                      buffers — an fsync cannot make user-space state durable, so the flush \
                      has to be whatever YOUR system's `really persist it` operation is; \
                      (2) the store path list is missing a file the engine needs to reopen; \
                      (3) the checker is looking in the wrong place."
                    .to_string(),
                violations: control.to_vec(),
            };
        }
    }

    // 2. Pre-workload state missing means the copy is not a usable store.
    if violations.iter().any(|v| v.kind == ViolationKind::Baseline) {
        return Finding::HarnessBroken {
            why: "the captured copy is missing state that already existed BEFORE the workload \
                  ran, so it is not a usable store. Make the setup state durable before the \
                  measured workload begins (write it, sync it, and record that sync)."
                .to_string(),
            violations: violations.to_vec(),
        };
    }

    if violations.is_empty() {
        return Finding::Survived;
    }

    // 3. A red under partial coverage is not a red. An unobserved sync might
    //    have saved the data, and guessing would libel the store.
    if let Coverage::Partial { unobserved } = coverage {
        if tier == Tier::PowerCut {
            return Finding::NotMeasured {
                why: format!(
                    "the durable set is incomplete, so this loss is not attributable: \
                     {unobserved}. powercut reports NotMeasured rather than a loss, because an \
                     unobserved sync may well have saved this data. Give the witness complete \
                     coverage — or configure the engine to perform no syncs of its own and do \
                     them yourself — before treating this as a finding."
                ),
            };
        }
    }

    Finding::Lost {
        violations: violations.to_vec(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::report::{Checker, Verdict};
    use crate::witness::SyncJournal;
    use crate::ScratchDir;

    fn read_back(crash: &CrashState) -> Option<Vec<u8>> {
        std::fs::read(crash.dir().join("app.log")).ok()
    }

    #[test]
    fn partial_coverage_downgrades_a_red_to_not_measured() {
        let store = ScratchDir::new("partial").unwrap();
        let db = store.path().join("app.log");
        let journal = SyncJournal::partial("the engine also syncs internally");

        let report = Scenario::new(store.path(), journal)
            .ack(|| std::fs::write(&db, b"acked").unwrap())
            .unwrap()
            .check(|crash, ()| {
                let mut c = Checker::new();
                c.survives("app.log", Some(b"acked".to_vec()), read_back(crash));
                c.finish()
            });

        assert_eq!(report.verdict(), Verdict::NotMeasured);
    }

    #[test]
    fn partial_coverage_still_allows_a_pass() {
        let store = ScratchDir::new("partial-pass").unwrap();
        let db = store.path().join("app.log");
        let journal = SyncJournal::partial("the engine also syncs internally");

        let report = Scenario::new(store.path(), journal.clone())
            .ack(|| {
                std::fs::write(&db, b"acked").unwrap();
                journal.sync_path(&db).unwrap();
            })
            .unwrap()
            .check(|crash, ()| {
                let mut c = Checker::new();
                c.survives("app.log", Some(b"acked".to_vec()), read_back(crash));
                c.finish()
            });

        assert_eq!(
            report.verdict(),
            Verdict::Survived,
            "a subset of the true durable set that still honours the ack is a sound pass"
        );
    }

    #[test]
    fn a_failing_teeth_control_is_harness_broken_not_lost() {
        let store = ScratchDir::new("broken").unwrap();
        let db = store.path().join("app.log");
        let journal = SyncJournal::complete();

        let report = Scenario::new(store.path(), journal.clone())
            .ack(|| std::fs::write(&db, b"acked").unwrap())
            .unwrap()
            .with_teeth_control(|| {
                journal.sync_path(&db)?;
                Ok(())
            })
            .unwrap()
            // A checker looking at the wrong file: the classic wrong-red.
            .check(|crash, ()| {
                let mut c = Checker::new();
                let wrong = std::fs::read(crash.dir().join("typo.log")).ok();
                c.survives("typo.log", Some(b"acked".to_vec()), wrong);
                c.finish()
            });

        assert_eq!(report.verdict(), Verdict::HarnessBroken);
    }
}