monetize-product 0.3.2

//! **The plan's building blocks every product's guardian writes the same way** (P4a; A22 W10,
//! A28 §4 "guardian helpers"): moved from Holger's guest (`holger-monetize-guest/src/guardian.rs`)
//! and Gunnar's (`gunnar-monetize-guest/src/guardian.rs`), which carried the same `order`,
//! `wait`, `recharge`, `restate`, `heard` and the same three bounds. What stays the product's:
//! its findings, which steps a remedy takes, and what its boxes' words mean.
//!
//! **The findings and plans both products answer line for line** (P4, lane mon-shared): the
//! both-lost and twin-lost findings, the main-lost finding over the guard's answer, the New
//! server rebuild, Review twin, Restore front and the `paired` look. Each guest calls these, so a
//! rule changed here changes both products' lifecycle at once. What stays each product's: how
//! it reads its twin (the guard's [`Twin`]), its own findings (Gunnar's `main_data_lost`), New
//! twin (Holger's box is told `become_twin`) and Recharge from twin (Holger's fresh-only rule).

use super::rebuild::{self, Rebuild, Twin};
use super::{role, BoxView, CloudStep, EstateView, Finding, Remedy, Step, StepState};

/// How long a laid box has to call home (cloud create + installer + first boot).
pub const HEARD_WITHIN_MS: u64 = 30 * 60_000;
/// How long the main has to show its new (or re-pointed) twin paired after adopting it.
pub const PAIRED_WITHIN_MS: u64 = 10 * 60_000;
/// How long a recharge may take to show the main's data whole again.
pub const RECHARGED_WITHIN_MS: u64 = 6 * 3_600_000;

/// A signed order of `kind` for the box in role `to`, its body's string values as given
/// (`$addr:<role>`, `$epoch`, `$principal` are filled in when the runner queues it).
pub fn order(to: &str, kind: &str, body: &[(&str, &str)]) -> Step {
    Step::Order {
        role: to.into(),
        kind: kind.into(),
        body: body
            .iter()
            .map(|(k, v)| (k.to_string(), v.to_string()))
            .collect(),
    }
}

/// Wait for the box in role `r` to show `what` within `within_ms`.
pub fn wait(r: &str, what: &str, within_ms: u64) -> Step {
    Step::Await {
        role: r.into(),
        what: what.into(),
        within_ms,
    }
}

/// The main filled from its twin (`recharge_from_twin`), at the case's fence epoch.
pub fn recharge() -> Step {
    order(
        super::role::MAIN,
        "recharge_from_twin",
        &[
            ("main", "$addr:main"),
            ("twin", "$addr:twin"),
            ("accept_lost", "false"),
            ("epoch", "$epoch"),
        ],
    )
}

/// The money back on a recharged main: monetize's books as a signed snapshot it pulls. The
/// twin carries the caps on a good day; only monetize can re-state a sold cap on the days it
/// does not.
pub fn restate() -> Step {
    order(
        super::role::MAIN,
        "restate_entitlements",
        &[("snapshot", "$entitlements")],
    )
}

/// One line of evidence about a box: when it was heard and what it said.
pub fn heard(v: &EstateView, r: &str) -> String {
    match v.get(r) {
        None => format!("the {r}: no box registered"),
        Some(b) if b.heard_unix_ms == 0 => {
            format!("the {r}: never heard since its key was registered")
        }
        Some(b) => {
            let words: Vec<String> = b.health.iter().map(|(k, v)| format!("{k}={v}")).collect();
            format!(
                "the {r}: last report {} s ago ({})",
                v.now_unix_ms.saturating_sub(b.heard_unix_ms) / 1000,
                words.join(" ")
            )
        }
    }
}

/// The health word `key` of the box in role `r` (empty: no such box, or it said none).
pub fn word<'a>(v: &'a EstateView, r: &str, key: &str) -> &'a str {
    v.get(r).map(|b| b.word(key)).unwrap_or("")
}

/// **Both boxes silent**: no remedy of its own; a human looks first.
pub fn both_lost(evidence: Vec<String>) -> Finding {
    Finding {
        code: "both_lost".into(),
        role: role::MAIN.into(),
        words: "the main and the twin are both silent: a human looks first (provider console, last receipts)".into(),
        evidence,
        remedies: Vec::new(),
        auto: None,
    }
}

/// **The main is lost** (silent, its twin up): `words` is the product's reading of it, the
/// guard's answer is said beside them, and auto-fix takes New server only where the guard says
/// lay again (the main GONE at the provider, a twin that can give it back and does not hear it).
pub fn main_lost(words: &str, guard: &Rebuild, evidence: Vec<String>) -> Finding {
    let auto = matches!(guard, Rebuild::LayAgain { .. });
    Finding {
        code: "main_lost".into(),
        role: role::MAIN.into(),
        words: format!("{words} — the guard: {}", guard.why()),
        evidence,
        remedies: vec![Remedy::NewServer, Remedy::RechargeFromTwin],
        auto: auto.then_some(Remedy::NewServer),
    }
}

/// **The twin is lost**: the main serves and its twin is silent, or the main says its twin is
/// `silent`/`lost`. `None` otherwise.
pub fn twin_lost(v: &EstateView, evidence: Vec<String>) -> Option<Finding> {
    let main_silent = v.silent(role::MAIN);
    let twin_silent = v.get(role::TWIN).is_some() && v.silent(role::TWIN);
    let main_says = !main_silent && matches!(word(v, role::MAIN, "twin"), "silent" | "lost");
    (!main_silent && (twin_silent || main_says)).then(|| Finding {
        code: "twin_lost".into(),
        role: role::TWIN.into(),
        words: "the twin is silent; the main serves alone".into(),
        evidence,
        remedies: vec![Remedy::NewTwin, Remedy::ReviewTwin],
        auto: Some(Remedy::NewTwin),
    })
}

/// **New server**: the main rebuilt from its twin, never promoted. The guard first (it gives
/// the old main back whole before the fence), then fence, lay, heard, recharge, recharged,
/// restate. `twin` is the product's reading of its twin.
pub fn new_server(v: &EstateView, twin: Option<Twin>) -> Result<Vec<Step>, String> {
    if v.silent(role::TWIN) {
        return Err("no twin is reporting: a new main would have nothing to recharge from".into());
    }
    rebuild::new_server(v, twin)?;
    Ok(vec![
        Step::Cloud(CloudStep::Fence {
            role: role::MAIN.into(),
        }),
        Step::Cloud(CloudStep::Lay {
            role: role::MAIN.into(),
        }),
        wait(role::MAIN, "heard", HEARD_WITHIN_MS),
        recharge(),
        wait(role::MAIN, "recharged", RECHARGED_WITHIN_MS),
        restate(),
    ])
}

/// **New twin** (shared, P4 lane v2-p4b; before it both guests carried the plan line for line):
/// the old twin fenced, a new one laid and heard, `told` first (what a product says to a fresh
/// twin — Holger's `become_twin`; Gunnar says nothing), then the main adopts it and is paired.
/// Refused while no main reports: the main comes first (New server).
pub fn new_twin(v: &EstateView, told: &[Step]) -> Result<Vec<Step>, String> {
    if v.silent(role::MAIN) {
        return Err(
            "no primary is reporting to adopt a new twin: the main comes first (New server)".into(),
        );
    }
    let mut steps = vec![
        Step::Cloud(CloudStep::Fence {
            role: role::TWIN.into(),
        }),
        Step::Cloud(CloudStep::Lay {
            role: role::TWIN.into(),
        }),
        wait(role::TWIN, "heard", HEARD_WITHIN_MS),
    ];
    steps.extend(told.iter().cloned());
    steps.push(order(
        role::MAIN,
        "adopt_twin",
        &[("twin", "$addr:twin"), ("principal", "$principal")],
    ));
    steps.push(wait(role::MAIN, "paired", PAIRED_WITHIN_MS));
    Ok(steps)
}

/// **Recharge from twin** (shared, P4 lane v2-p4b): the main pulls its data back from the twin.
/// Refused while the twin is silent (nothing to recharge from) or the main is silent (it cannot
/// pull its order: New server), then by the product's own `refuse` (Holger recharges only a
/// FRESH main; Gunnar refuses nothing more). A serving main is fenced first; then recharge,
/// `recharged` awaited, and the estate restated.
pub fn recharge_from_twin(
    v: &EstateView,
    refuse: impl FnOnce(&EstateView) -> Result<(), String>,
) -> Result<Vec<Step>, String> {
    if v.silent(role::TWIN) {
        return Err("no twin is reporting: nothing to recharge from".into());
    }
    if v.silent(role::MAIN) {
        return Err("the main is silent and cannot pull its order: press New server".into());
    }
    refuse(v)?;
    let mut steps = Vec::new();
    if word(v, role::MAIN, "serving") == "yes" {
        steps.push(order(
            role::MAIN,
            "fence_primary",
            &[("primary", "$addr:main"), ("epoch", "$epoch")],
        ));
    }
    steps.push(recharge());
    steps.push(wait(role::MAIN, "recharged", RECHARGED_WITHIN_MS));
    steps.push(restate());
    Ok(steps)
}

/// **Review twin**: the main reads its twin again (`review_twin`).
pub fn review_twin(v: &EstateView) -> Result<Vec<Step>, String> {
    if v.silent(role::MAIN) {
        return Err("the main is silent: nobody to review the twin".into());
    }
    Ok(vec![order(role::MAIN, "review_twin", &[])])
}

/// **Restore front**: the front reports nothing on Fleet; main and twin do, and they are the
/// proof that the estate the restored front comes back into stands.
pub fn restore_front(v: &EstateView) -> Result<Vec<Step>, String> {
    let (main_up, twin_up) = (!v.silent(role::MAIN), !v.silent(role::TWIN));
    if !main_up && !twin_up {
        return Err(
            "main and twin are silent: a front is restored only into an estate whose boxes stand"
                .into(),
        );
    }
    let heard = if main_up { role::MAIN } else { role::TWIN };
    Ok(vec![
        Step::Cloud(CloudStep::Restore {
            role: role::FRONT.into(),
        }),
        wait(heard, "heard", HEARD_WITHIN_MS),
    ])
}

/// **`recharged`, as the box in role `r` reports it** — GUNNAR's rule for every product (owner
/// 2026-11-03: Gunnar is the model): `data=ok` is met, whatever an older `recharge` word says;
/// otherwise `recharge=failed` fails the step; anything else waits.
pub fn observe_recharged(r: &str, b: &BoxView) -> StepState {
    match (b.word("data"), b.word("recharge")) {
        ("ok", _) => StepState::Met(format!("the {r} reports its data whole")),
        (_, "failed") => StepState::Failed(format!("the {r} reports the recharge failed")),
        (d, rc) => StepState::Waiting(format!("the {r} reports data={d} recharge={rc}")),
    }
}

/// **`paired`, as the box in role `r` reports it**: its `twin` word.
pub fn observe_paired(r: &str, b: &BoxView) -> StepState {
    match b.word("twin") {
        "paired" => StepState::Met(format!("the {r} reports its twin paired")),
        w => StepState::Waiting(format!("the {r} reports twin={w}")),
    }
}

#[cfg(test)]
mod tests {
    use super::super::{role, BoxView};
    use super::*;
    use std::collections::BTreeMap;

    fn view() -> EstateView {
        let mut boxes = BTreeMap::new();
        boxes.insert(
            "main".to_string(),
            BoxView {
                role: "main".into(),
                heard_unix_ms: 40_000,
                health: BTreeMap::from([
                    ("data".into(), "ok".into()),
                    ("serving".into(), "yes".into()),
                ]),
                ..BoxView::default()
            },
        );
        boxes.insert(
            "twin".to_string(),
            BoxView {
                role: "twin".into(),
                ..BoxView::default()
            },
        );
        EstateView {
            product: "p".into(),
            instance: "acme".into(),
            now_unix_ms: 100_000,
            silent_after_ms: 90_000,
            boxes,
        }
    }

    /// ★ P4 (lane v2-p4b): the New twin plan is one for every product; what a product tells a
    /// fresh twin goes in between heard and adopt_twin.
    #[test]
    fn a_new_twin_is_fenced_laid_heard_told_adopted_and_paired() {
        let v = view();
        let told = order(role::TWIN, "become_twin", &[("primary", "$addr:main")]);
        let steps = new_twin(&v, std::slice::from_ref(&told)).unwrap();
        assert_eq!(steps.len(), 6);
        assert_eq!(
            steps[0],
            Step::Cloud(CloudStep::Fence {
                role: "twin".into()
            })
        );
        assert_eq!(
            steps[1],
            Step::Cloud(CloudStep::Lay {
                role: "twin".into()
            })
        );
        assert_eq!(steps[2], wait(role::TWIN, "heard", HEARD_WITHIN_MS));
        assert_eq!(steps[3], told);
        assert_eq!(
            steps[4],
            order(
                role::MAIN,
                "adopt_twin",
                &[("twin", "$addr:twin"), ("principal", "$principal")]
            )
        );
        assert_eq!(steps[5], wait(role::MAIN, "paired", PAIRED_WITHIN_MS));
        assert_eq!(
            new_twin(&v, &[]).unwrap().len(),
            5,
            "a product that tells nothing"
        );
        let mut silent = view();
        silent.boxes.get_mut("main").unwrap().heard_unix_ms = 0;
        assert!(new_twin(&silent, &[])
            .unwrap_err()
            .contains("the main comes first"));
    }

    /// ★ P4 (lane v2-p4b): Recharge from twin is one plan; a product adds only its refusal.
    #[test]
    fn a_recharge_fences_a_serving_main_then_recharges_awaits_and_restates() {
        let mut v = view();
        v.boxes.get_mut("twin").unwrap().heard_unix_ms = 50_000;
        let steps = recharge_from_twin(&v, |_| Ok(())).unwrap();
        assert_eq!(
            steps,
            vec![
                order(
                    role::MAIN,
                    "fence_primary",
                    &[("primary", "$addr:main"), ("epoch", "$epoch")]
                ),
                recharge(),
                wait(role::MAIN, "recharged", RECHARGED_WITHIN_MS),
                restate(),
            ]
        );
        v.boxes
            .get_mut("main")
            .unwrap()
            .health
            .insert("serving".into(), "no".into());
        assert_eq!(
            recharge_from_twin(&v, |_| Ok(())).unwrap().len(),
            3,
            "a main that does not serve is not fenced"
        );
        let refused = recharge_from_twin(&v, |v| {
            Err(format!("data={}: not fresh", word(v, role::MAIN, "data")))
        })
        .unwrap_err();
        assert_eq!(
            refused, "data=ok: not fresh",
            "the product's own refusal, in its words"
        );
        let no_twin = view();
        assert!(recharge_from_twin(&no_twin, |_| Ok(()))
            .unwrap_err()
            .contains("no twin is reporting"));
        let mut silent_main = v.clone();
        silent_main.boxes.get_mut("main").unwrap().heard_unix_ms = 0;
        assert!(recharge_from_twin(&silent_main, |_| Ok(()))
            .unwrap_err()
            .contains("press New server"));
    }

    #[test]
    fn an_order_and_a_wait_are_the_steps_the_runner_reads() {
        assert_eq!(
            order(role::TWIN, "become_twin", &[("primary", "$addr:main")]),
            Step::Order {
                role: "twin".into(),
                kind: "become_twin".into(),
                body: BTreeMap::from([("primary".to_string(), "$addr:main".to_string())])
            }
        );
        assert_eq!(
            wait(role::MAIN, "paired", PAIRED_WITHIN_MS),
            Step::Await {
                role: "main".into(),
                what: "paired".into(),
                within_ms: 600_000
            }
        );
        assert_eq!(
            (HEARD_WITHIN_MS, RECHARGED_WITHIN_MS),
            (1_800_000, 21_600_000)
        );
    }

    #[test]
    fn the_recharge_and_the_restate_name_the_main_and_the_runners_fills() {
        let Step::Order { role, kind, body } = recharge() else {
            panic!("an order")
        };
        assert_eq!(
            (role.as_str(), kind.as_str()),
            ("main", "recharge_from_twin")
        );
        assert_eq!(
            body.iter()
                .map(|(k, v)| format!("{k}={v}"))
                .collect::<Vec<_>>(),
            [
                "accept_lost=false",
                "epoch=$epoch",
                "main=$addr:main",
                "twin=$addr:twin"
            ]
        );
        assert_eq!(
            restate(),
            order(
                "main",
                "restate_entitlements",
                &[("snapshot", "$entitlements")]
            )
        );
    }

    fn heard_at(r: &str, at: u64, words: &[(&str, &str)]) -> (String, BoxView) {
        (
            r.to_string(),
            BoxView {
                role: r.into(),
                heard_unix_ms: at,
                health: words
                    .iter()
                    .map(|(k, v)| (k.to_string(), v.to_string()))
                    .collect(),
                ..BoxView::default()
            },
        )
    }

    fn estate(boxes: Vec<(String, BoxView)>) -> EstateView {
        EstateView {
            product: "p".into(),
            instance: "acme".into(),
            now_unix_ms: 100_000,
            silent_after_ms: 90_000,
            boxes: boxes.into_iter().collect(),
        }
    }

    /// ★ P4 (lane mon-shared): the findings both guests wrote line for line, now one copy.
    #[test]
    fn the_shared_findings_are_both_lost_main_lost_and_twin_lost() {
        let b = both_lost(vec!["e".into()]);
        assert_eq!(
            (b.code.as_str(), b.role.as_str(), b.remedies.len(), b.auto),
            ("both_lost", "main", 0, None)
        );
        let guard = Rebuild::Refused {
            code: "x".into(),
            why: "not gone".into(),
        };
        let m = main_lost("the main is silent", &guard, Vec::new());
        assert_eq!(m.words, "the main is silent — the guard: not gone");
        assert_eq!(
            (m.remedies.clone(), m.auto),
            (vec![Remedy::NewServer, Remedy::RechargeFromTwin], None)
        );
        let serving = estate(vec![
            heard_at("main", 90_000, &[("serving", "yes")]),
            heard_at("twin", 1, &[]),
        ]);
        let t = twin_lost(&serving, Vec::new()).expect("a silent twin beside a serving main");
        assert_eq!(
            (t.code.as_str(), t.role.as_str(), t.auto),
            ("twin_lost", "twin", Some(Remedy::NewTwin))
        );
        let said = estate(vec![heard_at("main", 90_000, &[("twin", "lost")])]);
        assert!(
            twin_lost(&said, Vec::new()).is_some(),
            "the main says its twin is lost"
        );
        let whole = estate(vec![
            heard_at("main", 90_000, &[("twin", "paired")]),
            heard_at("twin", 90_000, &[]),
        ]);
        assert!(twin_lost(&whole, Vec::new()).is_none());
        let main_gone = estate(vec![heard_at("main", 1, &[]), heard_at("twin", 1, &[])]);
        assert!(
            twin_lost(&main_gone, Vec::new()).is_none(),
            "both silent is both_lost's"
        );
    }

    /// ★ P4 (lane mon-shared): the plans both guests wrote line for line, now one copy.
    #[test]
    fn the_shared_plans_are_new_server_review_twin_and_restore_front() {
        let both_up = estate(vec![
            heard_at("main", 90_000, &[]),
            heard_at("twin", 90_000, &[]),
        ]);
        let twin_only = estate(vec![
            heard_at("main", 1, &[]),
            heard_at("twin", 90_000, &[]),
        ]);
        let none_up = estate(vec![heard_at("main", 1, &[]), heard_at("twin", 1, &[])]);
        assert!(new_server(&none_up, None)
            .unwrap_err()
            .contains("no twin is reporting"));
        assert_eq!(
            review_twin(&both_up).unwrap(),
            vec![order("main", "review_twin", &[])]
        );
        assert!(review_twin(&twin_only)
            .unwrap_err()
            .contains("nobody to review"));
        assert_eq!(
            restore_front(&twin_only).unwrap(),
            vec![
                Step::Cloud(CloudStep::Restore {
                    role: "front".into()
                }),
                wait("twin", "heard", HEARD_WITHIN_MS)
            ]
        );
        assert_eq!(
            restore_front(&both_up).unwrap()[1],
            wait("main", "heard", HEARD_WITHIN_MS)
        );
        assert!(restore_front(&none_up).is_err());
        let (_, paired) = heard_at("main", 90_000, &[("twin", "paired")]);
        assert_eq!(
            observe_paired("main", &paired),
            StepState::Met("the main reports its twin paired".into())
        );
        let (_, whole) = heard_at("main", 90_000, &[("data", "ok"), ("recharge", "failed")]);
        assert_eq!(
            observe_recharged("main", &whole),
            StepState::Met("the main reports its data whole".into()),
            "Gunnar's rule: data=ok is met"
        );
        let (_, failed) = heard_at(
            "main",
            90_000,
            &[("data", "restoring"), ("recharge", "failed")],
        );
        assert!(matches!(
            observe_recharged("main", &failed),
            StepState::Failed(_)
        ));
        let (_, going) = heard_at("main", 90_000, &[("data", "restoring")]);
        assert!(matches!(
            observe_recharged("main", &going),
            StepState::Waiting(_)
        ));
        let (_, alone) = heard_at("main", 90_000, &[("twin", "silent")]);
        assert_eq!(
            observe_paired("main", &alone),
            StepState::Waiting("the main reports twin=silent".into())
        );
    }

    #[test]
    fn the_evidence_says_when_each_box_was_heard_and_what_it_said() {
        let v = view();
        assert_eq!(
            heard(&v, "main"),
            "the main: last report 60 s ago (data=ok serving=yes)"
        );
        assert_eq!(
            heard(&v, "twin"),
            "the twin: never heard since its key was registered"
        );
        assert_eq!(heard(&v, "front"), "the front: no box registered");
        assert_eq!(
            (
                word(&v, "main", "serving"),
                word(&v, "main", "twin"),
                word(&v, "front", "data")
            ),
            ("yes", "", "")
        );
    }
}