holger-fleet 4.0.0

//! **The order executor**: one verified order → one of holger's own verbs → one ack.
//!
//! The host hands a [`HolgerBox`] (holger-server-lib's `fleet_box`, a test's fake). Every
//! verb defaults to a refusal BY NAME, so a box that cannot do one says why instead of
//! the guardian waiting. The money is checked HERE as well as in the box: an entitlement,
//! a snapshot and a growth step are verified under the monetize key before the box is
//! asked to move anything.

use std::collections::BTreeMap;

use monetize_embed::fleet::{GrowGoAhead, OrderAck};
use monetize_product::pull::{AddPrincipal, GrowRoom, PurgeAnswer, PurgeTenant};
use monetize_product::EntitlementFact;

use crate::order::Order;

pub use crate::order::Recharge as RechargeAsk;

/// 512-byte sectors in a GiB.
pub const SECTORS_PER_GIB: u64 = 2_097_152;

/// What an entitlement did on the box.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Applied {
    /// Stored: the words say what.
    New(String),
    /// The box already holds this fact or a newer one: nothing moved.
    Held(String),
}

/// Where the grow machine stands, as a pulled growth reads it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum GrowState {
    Running(String),
    /// `/data` is off; the machine waits for the members to come back larger.
    WaitingMembers(String),
    Done(String),
    Failed(String),
}

/// **`Fleet.Ask{grow_set}`**: monetize sizes THIS box's own data disks.
pub trait GrowSet {
    fn grow_set(&mut self, gib: u64) -> Result<String, String>;
    /// **`Fleet.Ask{fetch}`**: the file a product order carries, PULLED from monetize (the ONE
    /// box never pushes into an estate; monetize MONETIZE-FIX).
    fn fetch(&mut self, f: &monetize_embed::fleet::Fetch) -> Result<Vec<u8>, String> {
        Err(format!(
            "this box has no pull channel to fetch {} on",
            f.param
        ))
    }
}

/// A caller with no channel to ask on.
pub struct NoAsk;
impl GrowSet for NoAsk {
    fn grow_set(&mut self, _: u64) -> Result<String, String> {
        Err("this box has no pull channel to ask monetize for its disks".into())
    }
}

/// **Holger's verbs, as an order reaches them.** Sync: the executor runs on a jera job's
/// blocking thread.
pub trait HolgerBox {
    /// This box serves as the primary (not a twin, not awaiting its recharge).
    fn serving(&self) -> bool;
    /// The box's health words for `Fleet.Report` (`serving`, `data`, `serves_at`, `twin`, `epoch`).
    fn health(&self) -> BTreeMap<String, String> {
        BTreeMap::new()
    }
    /// `monetize_product::pull::ProductReport` as JSON — the main that serves sends it.
    fn product_report(&self) -> Option<String> {
        None
    }
    /// A twin reviews itself on every report (G12): its review as JSON.
    fn own_review(&self) -> Option<String> {
        None
    }
    /// Store one VERIFIED fact, newer wins.
    fn entitle(&self, fact: &EntitlementFact) -> Result<Applied, String> {
        Err(format!(
            "this box holds no entitlements, so the fact for {} is not applied",
            fact.tenant.0
        ))
    }
    /// Purge one repository (or answer its purge's state).
    fn purge(&self, p: &PurgeTenant) -> Result<PurgeAnswer, String> {
        Err(format!(
            "this box purges no repositories, so {} is not purged",
            p.tenant
        ))
    }
    /// **A sold estate's resend sign-in link** (Gunnar's S10): mint a fresh first-login invite
    /// for the order's account and mail it to the order's address. Default: this box mints none.
    fn sign_in_again(&self, s: &crate::order::SignInAgain) -> Result<String, String> {
        let _ = (&s.to, &s.host, s.ttl_secs);
        Err(format!(
            "this box mints no sign-in links, so it cannot run sign_in_again for {}",
            s.account
        ))
    }
    /// **Keep the estate keys fed again** (monetize MAIL-RELAY + NIGHT-PUBREPO 2026-10-28): kept
    /// beside the feed the box was laid with, which they go over. Default: this box keeps none.
    fn keep_refeed(&self, r: &crate::order::Refeed) -> Result<String, String> {
        Err(format!(
            "this box keeps no estate keys fed again, so {} is not kept",
            r.keys.keys().cloned().collect::<Vec<_>>().join(", ")
        ))
    }
    /// **Keep one generic setting** (HOL-PUBSITE-FEED: a refeed's public-site switch, kept as
    /// [`crate::order::PUBLIC_SITE_SETTING`]): stored as an administrator's write to
    /// `/-/settings/<name>` is, audited, answering what it replaced. Default: this box keeps none.
    fn keep_setting(&self, name: &str, value: &[u8]) -> Result<String, String> {
        let _ = value;
        Err(format!("this box keeps no settings, so {name} is not kept"))
    }
    /// A paid signup's repository (`a.username`), bound to `a.signup` (idempotent); answers the
    /// namespace made. A name bound to another signup is refused starting `taken`.
    fn add_principal(&self, a: &AddPrincipal) -> Result<String, String> {
        Err(format!(
            "this box makes no repositories for a signup, so {} is not made",
            a.username
        ))
    }
    /// Keep the room wanted past the set and tell the operator.
    fn want_room(&self, g: &GrowRoom) -> Result<String, String> {
        Err(format!(
            "this box grows no data set, so {} bytes wanted cannot be kept",
            g.need_bytes
        ))
    }
    /// Do the box's data disks already carry `data_gib` a disk?
    fn grow_carried(&self, data_gib: u64) -> Result<bool, String> {
        Err(format!(
            "this box grows no data set, so it cannot grow to {data_gib} GiB"
        ))
    }
    /// The twin first (a main only): Ok once the twin holds main's whole set.
    fn twin_holds(&self, data_gib: u64) -> Result<String, String> {
        Err(format!(
            "this box cannot see whether its twin holds {data_gib} GiB a disk"
        ))
    }
    /// Start the growth to `data_gib` a disk under a one-time `nonce` (the step was verified).
    fn grow_start(&self, data_gib: u64, nonce: &str) -> Result<GrowState, String> {
        let _ = (data_gib, nonce);
        Err("this box grows no data set".into())
    }
    /// Wait for the grow machine to stop moving (past `waiting-members` when told).
    fn grow_settle(&self, past_waiting: bool) -> GrowState {
        let _ = past_waiting;
        GrowState::Failed("this box grows no data set".into())
    }
    /// Look at the members again (`give_up`: mount back what is there).
    fn grow_look_again(&self, give_up: bool) -> Result<(), String> {
        let _ = give_up;
        Err("this box grows no data set".into())
    }
    /// Fill `r.main` from `r.twin` (holger's recharge lane).
    fn recharge(&self, r: &RechargeAsk) -> Result<String, String> {
        Err(format!(
            "this box runs no recharge lane, so {} is not filled from {}",
            r.main, r.twin
        ))
    }
    /// Fence the old primary at `primary` (this box itself when it names it): it refuses
    /// every write from now on. Refused by name by default.
    fn fence(&self, primary: &str, epoch: u64) -> Result<String, String> {
        let _ = epoch;
        Err(format!("this box cannot fence the primary at {primary}: the guardian fences it at the cloud (the recharge's `fenced` proof)"))
    }
    /// Certify the twin at `twin` for this primary.
    fn adopt(&self, twin: &str, principal: &str) -> Result<String, String> {
        let _ = principal;
        Err(format!(
            "this box cannot certify the twin at {twin}: the front's certification does"
        ))
    }
    /// Take the twin role: follow `primary` (a FRESH box only, exactly once).
    /// **One of Holger's own order kinds** (`crate::order::holger_kind::PRODUCT`, queued by the
    /// operator on his site and pulled): its params as the operator typed them, its file pulled.
    /// `by` is the VERIFIED order it came in (its sequence and the monetize key it verified
    /// under): what the box's own audit names the write by. The answer is the ack's detail.
    fn product_order(
        &self,
        by: &OrderedBy,
        kind: &str,
        params: &BTreeMap<String, String>,
        file: Option<&[u8]>,
    ) -> Result<String, String> {
        let _ = (by, params, file);
        Err(format!("this box runs no {kind} order"))
    }
    fn become_twin(&self, primary: &str) -> Result<String, String> {
        Err(format!("this box cannot become the twin of {primary}"))
    }
    /// Review the twin (on a twin: itself), as JSON.
    fn review(&self) -> Result<serde_json::Value, String> {
        Err("this box cannot review a twin".into())
    }
}

/// What the executor knows about the order's context: the shared pump's own (P4a).
pub use monetize_estate_side::pump::Ctx;

/// **A `refeed`, kept group by group** (Gunnar's `gunnar-repair` `exec::refeed`): the keys kept
/// beside the feed first ([`HolgerBox::keep_refeed`]: the relay, the public repositories), then
/// each generic setting it sets ([`HolgerBox::keep_setting`]: the public-site switch). The first
/// refusal is the answer, saying what was kept before it; else every group's words.
fn refeed<B: HolgerBox + ?Sized>(b: &B, r: &crate::order::Refeed) -> Result<String, String> {
    let mut said: Vec<String> = Vec::new();
    let kept = r.kept_keys();
    if !kept.keys.is_empty() {
        said.push(b.keep_refeed(&kept)?);
    }
    for (name, value) in r.settings() {
        match b.keep_setting(name, &value) {
            Ok(words) => said.push(words),
            Err(why) if !said.is_empty() => {
                return Err(format!("{why} (kept before it: {})", said.join("; ")))
            }
            Err(why) => return Err(why),
        }
    }
    Ok(said.join("; "))
}

fn said(seq: u64, r: Result<String, String>) -> OrderAck {
    match r {
        Ok(detail) => OrderAck {
            seq,
            ok: true,
            detail,
            receipt_json: String::new(),
        },
        Err(detail) => OrderAck {
            seq,
            ok: false,
            detail,
            receipt_json: String::new(),
        },
    }
}

/// **The verified order a product order came in**: its sequence and the monetize key (hex) it
/// verified under. Made only by [`execute`] for an order the pump checked; never read off the
/// wire, so nothing a caller presents can be one.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OrderedBy {
    pub seq: u64,
    pub signer: String,
}

impl OrderedBy {
    /// The principal a box's audit names the order's write by: `fleet-order <seq> signed <key>`.
    pub fn principal(&self) -> String {
        format!("fleet-order {} signed {}", self.seq, self.signer)
    }
}

/// **Run one verified order** and say what came of it — the ack the next Report carries.
/// Never panics, never retries: the guardian reads the ack.
pub fn execute<B: HolgerBox + ?Sized>(
    b: &B,
    ctx: &Ctx,
    seq: u64,
    order: &Order,
    ask: &mut dyn GrowSet,
) -> OrderAck {
    let not_serving = || {
        Err::<String, String>("this box does not serve as the primary (a twin, or awaiting its recharge): nothing applied".into())
    };
    match order {
        Order::Entitlement(fact) => {
            if !b.serving() {
                return said(seq, not_serving());
            }
            if let Err(e) = monetize_embed::signing::verify_fact(fact, ctx.key) {
                return said(seq, Err(format!("the fact for {} does not verify under this box's monetize key ({e}): nothing moved", fact.tenant.0)));
            }
            said(
                seq,
                b.entitle(fact).map(|a| match a {
                    Applied::New(w) | Applied::Held(w) => w,
                }),
            )
        }
        Order::RestateEntitlements(snap) => {
            if !b.serving() {
                return said(seq, not_serving());
            }
            if let Err(e) = monetize_embed::signing::verify_snapshot(snap, ctx.key) {
                return said(seq, Err(format!("the snapshot does not verify under this box's monetize key ({e}): nothing restated")));
            }
            let (mut new, mut held) = (0usize, 0usize);
            for fact in &snap.facts {
                match b.entitle(fact) {
                    Ok(Applied::New(_)) => new += 1,
                    Ok(Applied::Held(_)) => held += 1,
                    Err(e) => {
                        return said(
                            seq,
                            Err(format!(
                                "restated {new} before {} was refused: {e}",
                                fact.tenant.0
                            )),
                        )
                    }
                }
            }
            said(seq, Ok(format!("restated {new} entitlement fact(s) ({held} already held as new or newer); nothing deleted")))
        }
        Order::PurgeTenant(p) => {
            if !b.serving() {
                return said(seq, not_serving());
            }
            match b.purge(p) {
                Ok(answer) => OrderAck {
                    seq,
                    ok: true,
                    detail: serde_json::to_string(&answer).unwrap_or_default(),
                    receipt_json: String::new(),
                },
                Err(e) => said(seq, Err(e)),
            }
        }
        Order::GrowRoom(g) => said(seq, b.want_room(g)),
        Order::AddPrincipal(a) => {
            if !b.serving() {
                return said(seq, not_serving());
            }
            said(seq, b.add_principal(a))
        }
        Order::SignInAgain(s) => {
            if !b.serving() {
                return said(seq, not_serving());
            }
            said(seq, b.sign_in_again(s))
        }
        // Any box the order is for keeps it (the pump checked the role): a twin that serves later
        // mails through it too.
        Order::Refeed(r) => said(seq, refeed(b, r)),
        Order::GrowGoAhead(g) => grow(b, ctx, seq, g, ask),
        Order::RechargeFromTwin(r) => {
            // Fence before recharge (Gunnar's order): the old primary the order names is
            // fenced first, and only a fence that held lets the recharge run beside it. A
            // guardian that fenced it AT THE CLOUD sends its proof (`fenced`) instead. With
            // neither, the box's recharge lane refuses `not_fenced` (A7 row 4: the shared
            // `nordisk_twin::fence::fence_gate`), so a bare order never makes two writers.
            let fenced = match r.fence.as_deref().map(str::trim) {
                Some(old) if old == r.main.trim() => {
                    return said(seq, Err(format!("order {seq} names the main {} as the primary to fence: a recharge never fences the box it fills", r.main)));
                }
                Some(old) => match b.fence(old, r.epoch) {
                    Ok(_) => Some(old.to_string()),
                    Err(e) => return said(seq, Err(format!("order {seq} fences {old} first and that fence did not hold ({e}): nothing was recharged"))),
                },
                None => None,
            };
            let prefix = match (&fenced, r.fenced.as_deref()) {
                (Some(f), _) => format!("fenced {f}; "),
                (None, Some(p)) => format!("cloud-fenced ({p}); "),
                (None, None) => String::new(),
            };
            said(
                seq,
                b.recharge(r)
                    .map(|w| format!("{prefix}{w}"))
                    .map_err(|e| format!("{prefix}{e}")),
            )
        }
        Order::FencePrimary { primary, epoch } => said(seq, b.fence(primary, *epoch)),
        Order::AdoptTwin { twin, principal } => said(seq, b.adopt(twin, principal)),
        Order::BecomeTwin { primary } => said(seq, b.become_twin(primary)),
        Order::Product { kind, body } => {
            if !b.serving() {
                return said(seq, not_serving());
            }
            let file = match &body.fetch {
                None => None,
                Some(f) => match ask.fetch(f) {
                    Ok(bytes) => Some(bytes),
                    Err(e) => {
                        return said(
                            seq,
                            Err(format!(
                                "order {seq} ({kind}): its file {} was not pulled: {e}",
                                f.param
                            )),
                        )
                    }
                },
            };
            let by = OrderedBy {
                seq,
                signer: crate::hex(&ctx.key.to_bytes()),
            };
            said(
                seq,
                b.product_order(&by, kind, &body.params, file.as_deref()),
            )
        }
        Order::ReviewTwin => match b.review() {
            Ok(v) => OrderAck {
                seq,
                ok: true,
                detail: v
                    .get("verdict")
                    .and_then(|w| w.as_str())
                    .unwrap_or("reviewed")
                    .to_string(),
                receipt_json: v.to_string(),
            },
            Err(e) => said(seq, Err(e)),
        },
    }
}

/// **A pulled `grow_go_ahead`** (G16-d): verify monetize's step for THIS estate and role,
/// the twin first, drain and unmount, ask monetize to size this box's own disks, look
/// again, remount. A growth already at size is acked ok with nothing moved.
fn grow<B: HolgerBox + ?Sized>(
    b: &B,
    ctx: &Ctx,
    seq: u64,
    g: &GrowGoAhead,
    ask: &mut dyn GrowSet,
) -> OrderAck {
    let fail = |w: String| said(seq, Err(w));
    let step = match monetize_embed::signing::verify_estate_growth(g.go_ahead.as_bytes(), ctx.key) {
        Ok(s) => s,
        Err(e) => {
            return fail(format!(
                "the paid step does not verify under this box's monetize key ({e}): nothing moved"
            ))
        }
    };
    if step.estate != ctx.me.instance {
        return fail(format!(
            "the paid step is estate {}'s; this box pulls as {}: nothing moved",
            step.estate, ctx.me.instance
        ));
    }
    if step.until_unix_ms <= ctx.now_unix_ms as i64 {
        return fail(format!(
            "the paid step was good until {} (unix ms): its period is over, nothing moved",
            step.until_unix_ms
        ));
    }
    let twin = ctx.me.role == "twin";
    let main_gib = step.main_data_gib_per_member;
    if !twin && g.data_gib != main_gib {
        return fail(format!("the order sizes main's disks to {} GiB, the paid step to {main_gib} GiB: nothing moved", g.data_gib));
    }
    if twin && g.data_gib < main_gib {
        return fail(format!("the order sizes the twin's disk to {} GiB, under main's {main_gib} GiB a disk: nothing moved", g.data_gib));
    }
    match b.grow_carried(g.data_gib) {
        Ok(true) => {
            return said(
                seq,
                Ok(format!(
                    "already {} GiB a data disk: nothing moved",
                    g.data_gib
                )),
            )
        }
        Ok(false) => {}
        Err(e) => return fail(e),
    }
    if !twin {
        if let Err(e) = b.twin_holds(g.data_gib) {
            return fail(e);
        }
    }
    match b.grow_start(g.data_gib, &format!("pull-{seq}-{}", step.reference)) {
        Ok(GrowState::Done(d)) => return said(seq, Ok(d)),
        Ok(GrowState::Failed(f)) => return fail(f),
        Ok(_) => {}
        Err(e) => return fail(e),
    }
    match b.grow_settle(false) {
        GrowState::WaitingMembers(_) => {}
        GrowState::Done(d) => return said(seq, Ok(d)),
        GrowState::Failed(f) | GrowState::Running(f) => return fail(f),
    }
    let asked = ask.grow_set(g.data_gib);
    if let Err(e) = b.grow_look_again(asked.is_err()) {
        return fail(format!("the grow machine would not look again: {e}"));
    }
    let end = b.grow_settle(true);
    match (asked, end) {
        (Ok(words), GrowState::Done(d)) => said(seq, Ok(format!("{d} (monetize: {words})"))),
        (
            Ok(words),
            GrowState::Failed(f) | GrowState::Running(f) | GrowState::WaitingMembers(f),
        ) => fail(format!("{f} (monetize: {words})")),
        (
            Err(e),
            GrowState::Done(s)
            | GrowState::Failed(s)
            | GrowState::Running(s)
            | GrowState::WaitingMembers(s),
        ) => fail(format!(
            "monetize did not size this box's disks ({e}); /data: {s}"
        )),
    }
}

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use crate::order::tests::{me, monetize_key, MONETIZE_SEED};
    use ed25519_dalek::Signer as _;
    use monetize_product::{State, TenantId};
    use std::cell::RefCell;

    /// A Holger box in memory: what it holds and every verb it was asked.
    #[derive(Default)]
    pub struct Fake {
        pub twin: bool,
        pub facts: RefCell<BTreeMap<String, EntitlementFact>>,
        pub did: RefCell<Vec<String>>,
        pub carried: bool,
        pub settle: RefCell<Vec<GrowState>>,
    }

    impl HolgerBox for Fake {
        fn serving(&self) -> bool {
            !self.twin
        }
        fn entitle(&self, fact: &EntitlementFact) -> Result<Applied, String> {
            let mut f = self.facts.borrow_mut();
            if f.get(&fact.tenant.0)
                .is_some_and(|h| h.issued_unix_ms >= fact.issued_unix_ms)
            {
                return Ok(Applied::Held(format!("{}: held", fact.tenant.0)));
            }
            f.insert(fact.tenant.0.clone(), fact.clone());
            Ok(Applied::New(format!(
                "{} is {:?}",
                fact.tenant.0, fact.state
            )))
        }
        fn purge(&self, p: &PurgeTenant) -> Result<PurgeAnswer, String> {
            self.did.borrow_mut().push(format!("purge {}", p.tenant));
            Ok(PurgeAnswer {
                job: format!("purge:{}", p.tenant),
                detail: "3 artifacts".into(),
                ..Default::default()
            })
        }
        fn want_room(&self, g: &GrowRoom) -> Result<String, String> {
            self.did.borrow_mut().push(format!("room {}", g.need_bytes));
            Ok("kept".into())
        }
        fn sign_in_again(&self, s: &crate::order::SignInAgain) -> Result<String, String> {
            self.did.borrow_mut().push(format!(
                "sign-in {} {} {} {}",
                s.account, s.to, s.host, s.ttl_secs
            ));
            Ok(format!(
                "a new sign-in link for {} was mailed to {}",
                s.account, s.to
            ))
        }
        fn keep_refeed(&self, r: &crate::order::Refeed) -> Result<String, String> {
            self.did.borrow_mut().push(format!(
                "refeed {}",
                r.keys.keys().cloned().collect::<Vec<_>>().join(",")
            ));
            Ok(format!(
                "kept {}",
                r.keys.keys().cloned().collect::<Vec<_>>().join(", ")
            ))
        }
        fn keep_setting(&self, name: &str, value: &[u8]) -> Result<String, String> {
            let v = String::from_utf8_lossy(value).into_owned();
            self.did.borrow_mut().push(format!("setting {name} {v}"));
            Ok(format!("{name} is {v}"))
        }
        fn add_principal(&self, a: &AddPrincipal) -> Result<String, String> {
            self.did
                .borrow_mut()
                .push(format!("signup {} {}", a.username, a.signup));
            if a.username == "theirs" {
                Err("taken: theirs is another signup's".into())
            } else {
                Ok(a.username.clone())
            }
        }
        fn grow_carried(&self, _: u64) -> Result<bool, String> {
            Ok(self.carried)
        }
        fn twin_holds(&self, gib: u64) -> Result<String, String> {
            self.did.borrow_mut().push(format!("twin_holds {gib}"));
            Ok("the twin holds it".into())
        }
        fn grow_start(&self, gib: u64, nonce: &str) -> Result<GrowState, String> {
            self.did.borrow_mut().push(format!("start {gib} {nonce}"));
            Ok(GrowState::Running("draining".into()))
        }
        fn grow_settle(&self, past: bool) -> GrowState {
            self.did.borrow_mut().push(format!("settle {past}"));
            let mut s = self.settle.borrow_mut();
            if s.is_empty() {
                if past {
                    GrowState::Done("grown".into())
                } else {
                    GrowState::WaitingMembers("waiting-members".into())
                }
            } else {
                s.remove(0)
            }
        }
        fn grow_look_again(&self, give_up: bool) -> Result<(), String> {
            self.did.borrow_mut().push(format!("look_again {give_up}"));
            Ok(())
        }
        fn recharge(&self, r: &RechargeAsk) -> Result<String, String> {
            self.did.borrow_mut().push(format!(
                "recharge {} from {} lost={}",
                r.main, r.twin, r.accept_lost
            ));
            Ok("recharged".into())
        }
        fn fence(&self, primary: &str, epoch: u64) -> Result<String, String> {
            self.did
                .borrow_mut()
                .push(format!("fence {primary} {epoch}"));
            if primary.starts_with("stuck") {
                Err(format!("{primary} answers nothing"))
            } else {
                Ok(format!("{primary} is fenced at epoch {epoch}"))
            }
        }
        fn adopt(&self, twin: &str, principal: &str) -> Result<String, String> {
            self.did
                .borrow_mut()
                .push(format!("adopt {twin} {principal}"));
            Ok("follows".into())
        }
        fn review(&self) -> Result<serde_json::Value, String> {
            Ok(serde_json::json!({"verdict": "the twin is whole", "complete": true}))
        }
        fn become_twin(&self, primary: &str) -> Result<String, String> {
            self.did.borrow_mut().push(format!("become_twin {primary}"));
            Ok("elected".into())
        }
        fn product_order(
            &self,
            by: &OrderedBy,
            kind: &str,
            params: &BTreeMap<String, String>,
            file: Option<&[u8]>,
        ) -> Result<String, String> {
            self.did.borrow_mut().push(format!(
                "{kind} {params:?} {:?} by {}",
                file.map(<[u8]>::len),
                by.seq
            ));
            Ok(format!("{{\"ran\":\"{kind}\"}}"))
        }
    }

    pub fn fact(repo: &str, issued: u64, signer: [u8; 32]) -> EntitlementFact {
        let mut f = EntitlementFact {
            tenant: TenantId(repo.into()),
            plan: "o-1".into(),
            state: State::Paid,
            paid_until_unix_ms: Some(99_000_000),
            caps: Default::default(),
            source: "payment:stripe:o-1".into(),
            signature: Vec::new(),
            issued_unix_ms: (issued != 0).then_some(issued),
            issued_signature: Vec::new(),
        };
        let key = ed25519_dalek::SigningKey::from_bytes(&signer);
        f.signature = key
            .sign(&monetize_embed::signing::fact_message(&f))
            .to_bytes()
            .to_vec();
        if issued != 0 {
            f.issued_signature = key
                .sign(&monetize_embed::signing::fact_message_issued(&f))
                .to_bytes()
                .to_vec();
        }
        f
    }

    fn run(b: &dyn HolgerBox, role: &str, o: Order) -> OrderAck {
        let (m, k) = (me(role), monetize_key());
        execute(
            b,
            &Ctx {
                me: &m,
                key: &k,
                now_unix_ms: 10_000,
            },
            5,
            &o,
            &mut NoAsk,
        )
    }

    struct Asked(Vec<u64>, bool);
    impl GrowSet for Asked {
        fn grow_set(&mut self, gib: u64) -> Result<String, String> {
            self.0.push(gib);
            if self.1 {
                Ok("resized".into())
            } else {
                Err("no answer".into())
            }
        }
    }

    fn growth(estate: &str, gib: u64, until: i64) -> String {
        let mut g = monetize_embed::signing::EstateGrowth {
            v: 1,
            signer: "shop".into(),
            estate: estate.into(),
            main_data_gib_per_member: gib,
            until_unix_ms: until,
            reference: "o-7".into(),
            issued_unix_ms: 1,
            signature: String::new(),
        };
        let sig = ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED)
            .sign(&monetize_embed::signing::estate_growth_message(&g));
        g.signature = monetize_embed::signing::base64_encode(&sig.to_bytes());
        serde_json::to_string(&g).unwrap()
    }

    #[test]
    fn entitlement_is_applied_newer_wins_and_a_forgery_moves_nothing() {
        let b = Fake::default();
        let a = run(
            &b,
            "main",
            Order::Entitlement(fact("libs", 5, MONETIZE_SEED)),
        );
        assert!(a.ok, "{a:?}");
        assert!(b.facts.borrow().contains_key("libs"));
        let again = run(
            &b,
            "main",
            Order::Entitlement(fact("libs", 4, MONETIZE_SEED)),
        );
        assert!(
            again.ok && again.detail.contains("held"),
            "an older fact is held, not refused: {again:?}"
        );
        let forged = run(&b, "main", Order::Entitlement(fact("other", 6, [9u8; 32])));
        assert!(
            !forged.ok && forged.detail.contains("does not verify"),
            "{forged:?}"
        );
        assert!(!b.facts.borrow().contains_key("other"));
        let twin = Fake {
            twin: true,
            ..Default::default()
        };
        let refused = run(
            &twin,
            "twin",
            Order::Entitlement(fact("libs", 5, MONETIZE_SEED)),
        );
        assert!(
            !refused.ok && refused.detail.contains("does not serve"),
            "{refused:?}"
        );
    }

    /// ★ **A resend sign-in link reaches the box that serves** (A16, Gunnar's S10): the order's
    /// words go to the host as they are; a twin mints nothing; a box that cannot mint says so.
    #[test]
    fn a_sign_in_again_reaches_the_serving_box_and_a_twin_mints_nothing() {
        let s = crate::order::SignInAgain {
            account: "acme".into(),
            to: "buyer@mock.invalid".into(),
            host: "acme.holger.mock.invalid".into(),
            ttl_secs: 86_400,
        };
        let b = Fake::default();
        let a = run(&b, "main", Order::SignInAgain(s.clone()));
        assert!(
            a.ok && a.detail.contains("mailed to buyer@mock.invalid"),
            "{a:?}"
        );
        assert_eq!(
            b.did.borrow().as_slice(),
            ["sign-in acme buyer@mock.invalid acme.holger.mock.invalid 86400"]
        );
        let twin = Fake {
            twin: true,
            ..Default::default()
        };
        let refused = run(&twin, "twin", Order::SignInAgain(s.clone()));
        assert!(
            !refused.ok
                && refused.detail.contains("does not serve")
                && twin.did.borrow().is_empty(),
            "{refused:?}"
        );
        struct Bare;
        impl HolgerBox for Bare {
            fn serving(&self) -> bool {
                true
            }
        }
        let none = run(&Bare, "main", Order::SignInAgain(s));
        assert!(
            !none.ok && none.detail.contains("mints no sign-in link"),
            "{none:?}"
        );
    }

    /// ★ **HOL-PUBSITE-FEED (Gunnar's GRS-PUBLIC): a refeed's public-site switch is kept as the
    /// box's generic setting** `console.public-site` (where an administrator's flip is kept), after
    /// the keys kept beside the feed; the switch alone reaches only the setting. A box that keeps
    /// no settings refuses by name and says what it kept before.
    #[test]
    fn a_refeeds_public_site_switch_is_kept_as_the_consoles_setting() {
        let keys = |kv: &[(&str, &str)]| crate::order::Refeed {
            keys: kv
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect(),
        };
        let b = Fake::default();
        let a = run(
            &b,
            "main",
            Order::Refeed(keys(&[
                ("console_public_site", "on"),
                ("public_repositories", "crates"),
            ])),
        );
        assert!(a.ok && a.detail.contains("console.public-site"), "{a:?}");
        assert_eq!(
            b.did.borrow().as_slice(),
            [
                "refeed public_repositories",
                "setting console.public-site on"
            ]
        );
        let b = Fake::default();
        let a = run(
            &b,
            "twin",
            Order::Refeed(keys(&[("console_public_site", "off")])),
        );
        assert!(a.ok, "{a:?}");
        assert_eq!(
            b.did.borrow().as_slice(),
            ["setting console.public-site off"],
            "the switch alone keeps no key beside the feed"
        );
        struct KeepsKeysOnly;
        impl HolgerBox for KeepsKeysOnly {
            fn serving(&self) -> bool {
                true
            }
            fn keep_refeed(&self, _: &crate::order::Refeed) -> Result<String, String> {
                Ok("kept public_repositories".into())
            }
        }
        let none = run(
            &KeepsKeysOnly,
            "main",
            Order::Refeed(keys(&[
                ("console_public_site", "on"),
                ("public_repositories", "crates"),
            ])),
        );
        assert!(
            !none.ok
                && none.detail.contains("keeps no settings")
                && none
                    .detail
                    .contains("kept before it: kept public_repositories"),
            "{none:?}"
        );
    }

    /// ★ **MAIL-RELAY: a refeed reaches the box's keep**; a box that keeps none says so.
    #[test]
    fn a_refeed_reaches_the_boxes_keep_and_a_box_that_keeps_none_refuses() {
        let r = crate::order::Refeed {
            keys: [
                ("secret.smtp_url", "smtps://u:tok@relay.mock.invalid:465"),
                ("secret.smtp_from", "holger@mock.invalid"),
                ("public_repositories", "crates"),
            ]
            .iter()
            .map(|(k, v)| (k.to_string(), v.to_string()))
            .collect(),
        };
        let b = Fake::default();
        let a = run(&b, "main", Order::Refeed(r.clone()));
        assert!(
            a.ok && a.detail.contains("public_repositories") && !a.detail.contains("tok@"),
            "{a:?}"
        );
        assert_eq!(
            b.did.borrow().as_slice(),
            ["refeed public_repositories,secret.smtp_from,secret.smtp_url"]
        );
        struct Bare;
        impl HolgerBox for Bare {
            fn serving(&self) -> bool {
                true
            }
        }
        let none = run(&Bare, "main", Order::Refeed(r));
        assert!(
            !none.ok && none.detail.contains("keeps no estate keys"),
            "{none:?}"
        );
    }

    #[test]
    fn restate_entitlements_verifies_the_snapshot_and_counts_new_and_held() {
        let b = Fake::default();
        run(&b, "main", Order::Entitlement(fact("a", 5, MONETIZE_SEED)));
        let facts = vec![fact("a", 5, MONETIZE_SEED), fact("b", 5, MONETIZE_SEED)];
        let mut snap = monetize_embed::Snapshot {
            issued_unix_ms: 7,
            facts,
            signature: Vec::new(),
        };
        snap.signature = ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED)
            .sign(&monetize_embed::signing::snapshot_message(
                snap.issued_unix_ms,
                &snap.facts,
            ))
            .to_bytes()
            .to_vec();
        let a = run(&b, "main", Order::RestateEntitlements(snap.clone()));
        assert!(
            a.ok && a.detail.contains("restated 1") && a.detail.contains("1 already held"),
            "{a:?}"
        );
        snap.signature = vec![0; 64];
        let bad = run(&b, "main", Order::RestateEntitlements(snap));
        assert!(!bad.ok && bad.detail.contains("does not verify"), "{bad:?}");
    }

    #[test]
    fn purge_tenant_answers_monetizes_purge_answer() {
        let b = Fake::default();
        let a = run(
            &b,
            "main",
            Order::PurgeTenant(PurgeTenant {
                tenant: "libs".into(),
                reason: "nonpay".into(),
                job: String::new(),
            }),
        );
        assert!(a.ok, "{a:?}");
        let answer: PurgeAnswer = serde_json::from_str(&a.detail).unwrap();
        assert_eq!(answer.job, "purge:libs");
        assert_eq!(b.did.borrow().as_slice(), ["purge libs"]);
    }

    #[test]
    fn add_principal_makes_a_repository_on_a_plan_and_says_taken_by_name() {
        let b = Fake::default();
        let a = run(
            &b,
            "main",
            Order::AddPrincipal(AddPrincipal {
                username: "crates".into(),
                kind: "org".into(),
                signup: "s-1".into(),
            }),
        );
        assert!(
            a.ok && a.detail == "crates",
            "the namespace made is the repository: {a:?}"
        );
        let t = run(
            &b,
            "main",
            Order::AddPrincipal(AddPrincipal {
                username: "theirs".into(),
                kind: "org".into(),
                signup: "s-2".into(),
            }),
        );
        assert!(!t.ok && t.detail.starts_with("taken"), "{t:?}");
        let twin = Fake {
            twin: true,
            ..Default::default()
        };
        assert!(
            !run(
                &twin,
                "twin",
                Order::AddPrincipal(AddPrincipal {
                    username: "crates".into(),
                    kind: "org".into(),
                    signup: "s-1".into()
                })
            )
            .ok,
            "only the main that serves"
        );
    }

    #[test]
    fn grow_room_is_kept_for_the_operator() {
        let b = Fake::default();
        let a = run(
            &b,
            "main",
            Order::GrowRoom(GrowRoom {
                need_bytes: 1 << 30,
                reason: "libs".into(),
            }),
        );
        assert!(a.ok, "{a:?}");
        assert_eq!(b.did.borrow().as_slice(), ["room 1073741824"]);
    }

    #[test]
    fn grow_go_ahead_runs_twin_first_asks_monetize_and_remounts() {
        let b = Fake::default();
        let (m, k) = (me("main"), monetize_key());
        let mut ask = Asked(Vec::new(), true);
        let o = Order::GrowGoAhead(GrowGoAhead {
            go_ahead: growth("acme", 120, 99_000),
            data_gib: 120,
        });
        let a = execute(
            &b,
            &Ctx {
                me: &m,
                key: &k,
                now_unix_ms: 10_000,
            },
            6,
            &o,
            &mut ask,
        );
        assert!(a.ok, "{a:?}");
        assert_eq!(
            b.did.borrow().as_slice(),
            [
                "twin_holds 120",
                "start 120 pull-6-o-7",
                "settle false",
                "look_again false",
                "settle true"
            ]
        );
        assert_eq!(ask.0, [120]);

        // Monetize does not answer: /data is mounted back and the ack says why.
        let b = Fake::default();
        let mut deaf = Asked(Vec::new(), false);
        b.settle.borrow_mut().extend([
            GrowState::WaitingMembers("w".into()),
            GrowState::Failed("mounted back".into()),
        ]);
        let a = execute(
            &b,
            &Ctx {
                me: &m,
                key: &k,
                now_unix_ms: 10_000,
            },
            7,
            &o,
            &mut deaf,
        );
        assert!(
            !a.ok && a.detail.contains("no answer") && a.detail.contains("mounted back"),
            "{a:?}"
        );
        assert!(b.did.borrow().contains(&"look_again true".to_string()));
    }

    #[test]
    fn grow_go_ahead_refuses_another_estate_a_lapsed_step_and_a_forgery_and_skips_a_done_one() {
        let b = Fake::default();
        for (go, why) in [
            (growth("other", 120, 99_000), "estate other"),
            (growth("acme", 120, 5_000), "period is over"),
            (
                growth("acme", 120, 99_000).replace("\"o-7\"", "\"o-8\""),
                "does not verify",
            ),
        ] {
            let a = run(
                &b,
                "main",
                Order::GrowGoAhead(GrowGoAhead {
                    go_ahead: go,
                    data_gib: 120,
                }),
            );
            assert!(!a.ok && a.detail.contains(why), "{why}: {a:?}");
        }
        assert!(b.did.borrow().is_empty(), "nothing moved");
        let done = Fake {
            carried: true,
            ..Default::default()
        };
        let a = run(
            &done,
            "main",
            Order::GrowGoAhead(GrowGoAhead {
                go_ahead: growth("acme", 120, 99_000),
                data_gib: 120,
            }),
        );
        assert!(a.ok && a.detail.contains("nothing moved"), "{a:?}");
        let twin = Fake {
            twin: true,
            ..Default::default()
        };
        let a = run(
            &twin,
            "twin",
            Order::GrowGoAhead(GrowGoAhead {
                go_ahead: growth("acme", 120, 99_000),
                data_gib: 240,
            }),
        );
        assert!(a.detail.contains("no pull channel"), "{a:?}");
        assert!(
            !twin
                .did
                .borrow()
                .iter()
                .any(|d| d.starts_with("twin_holds")),
            "the twin grows first, asking no twin: {a:?}"
        );
    }

    #[test]
    fn recharge_from_twin_runs_the_lane_on_a_cloud_fence_proof() {
        let b = Fake::default();
        let r = RechargeAsk {
            main: "main:1".into(),
            twin: "twin:1".into(),
            accept_lost: true,
            fenced: Some("upcloud: gone".into()),
            ..Default::default()
        };
        let a = run(&b, "twin", Order::RechargeFromTwin(r));
        assert!(a.ok && a.detail.starts_with("cloud-fenced"), "{a:?}");
        assert_eq!(
            b.did.borrow().as_slice(),
            ["recharge main:1 from twin:1 lost=true"]
        );
    }

    /// ★ **Fence before recharge, by the pulled order** (M2, Gunnar's pattern): the old
    /// primary the order names is fenced first, and only a fence that held lets the
    /// recharge run; a fence that did not hold runs nothing; the main it fills is never
    /// the box it fences.
    #[test]
    fn recharge_from_twin_fences_the_old_primary_first() {
        let b = Fake::default();
        let r = RechargeAsk {
            main: "main:2".into(),
            twin: "twin:1".into(),
            accept_lost: false,
            epoch: 4,
            fence: Some("old:1".into()),
            ..Default::default()
        };
        let a = run(&b, "twin", Order::RechargeFromTwin(r.clone()));
        assert!(a.ok && a.detail.starts_with("fenced old:1; "), "{a:?}");
        assert_eq!(
            b.did.borrow().as_slice(),
            ["fence old:1 4", "recharge main:2 from twin:1 lost=false"]
        );

        let stuck = Fake::default();
        let a = run(
            &stuck,
            "twin",
            Order::RechargeFromTwin(RechargeAsk {
                fence: Some("stuck:1".into()),
                ..r.clone()
            }),
        );
        assert!(
            !a.ok
                && a.detail.contains("stuck:1 answers nothing")
                && a.detail.contains("nothing was recharged"),
            "{a:?}"
        );
        assert_eq!(
            stuck.did.borrow().as_slice(),
            ["fence stuck:1 4"],
            "no recharge beside an unfenced primary"
        );

        let same = Fake::default();
        let a = run(
            &same,
            "twin",
            Order::RechargeFromTwin(RechargeAsk {
                fence: Some("main:2".into()),
                ..r
            }),
        );
        assert!(
            !a.ok && a.detail.contains("never fences the box it fills"),
            "{a:?}"
        );
        assert!(same.did.borrow().is_empty(), "nothing ran");
    }

    #[test]
    fn become_twin_runs_holgers_election() {
        let b = Fake {
            twin: true,
            ..Default::default()
        };
        assert!(
            run(
                &b,
                "twin",
                Order::BecomeTwin {
                    primary: "main:1".into()
                }
            )
            .ok
        );
        assert_eq!(b.did.borrow().as_slice(), ["become_twin main:1"]);
        struct Bare;
        impl HolgerBox for Bare {
            fn serving(&self) -> bool {
                false
            }
        }
        let none = run(
            &Bare,
            "twin",
            Order::BecomeTwin {
                primary: "main:1".into(),
            },
        );
        assert!(
            !none.ok && none.detail.contains("cannot become"),
            "the default refuses by name: {none:?}"
        );
    }

    #[test]
    fn fence_primary_runs_the_boxes_fence_adopt_and_review_run() {
        let b = Fake::default();
        let a = run(
            &b,
            "twin",
            Order::FencePrimary {
                primary: "old:1".into(),
                epoch: 3,
            },
        );
        assert!(
            a.ok && b.did.borrow().contains(&"fence old:1 3".to_string()),
            "{a:?}"
        );
        struct Bare;
        impl HolgerBox for Bare {
            fn serving(&self) -> bool {
                false
            }
        }
        let none = run(
            &Bare,
            "twin",
            Order::FencePrimary {
                primary: "old:1".into(),
                epoch: 3,
            },
        );
        assert!(
            !none.ok && none.detail.contains("cannot fence"),
            "the default refuses by name: {none:?}"
        );
        let a = run(
            &b,
            "main",
            Order::AdoptTwin {
                twin: "twin:1".into(),
                principal: "twin".into(),
            },
        );
        assert!(
            a.ok && b.did.borrow().contains(&"adopt twin:1 twin".to_string()),
            "{a:?}"
        );
        let a = run(&b, "twin", Order::ReviewTwin);
        assert!(
            a.ok && a.detail == "the twin is whole" && a.receipt_json.contains("complete"),
            "{a:?}"
        );
    }

    /// ★ MONETIZE-FIX: **one of Holger's own kinds runs through the box's verb** with its params;
    /// its file is PULLED first (a fetch that fails is said, and nothing runs); a twin runs none.
    #[test]
    fn a_product_order_runs_with_its_params_and_its_pulled_file() {
        use monetize_embed::fleet::{Fetch, ProductOrder};
        struct Shop(Result<Vec<u8>, String>);
        impl GrowSet for Shop {
            fn grow_set(&mut self, _: u64) -> Result<String, String> {
                Err("no".into())
            }
            fn fetch(&mut self, _: &Fetch) -> Result<Vec<u8>, String> {
                self.0.clone()
            }
        }
        let (m, k) = (me("main"), monetize_key());
        let ctx = Ctx {
            me: &m,
            key: &k,
            now_unix_ms: 10_000,
        };
        let b = Fake::default();
        let params: BTreeMap<String, String> = [("name".to_string(), "lab-1".to_string())].into();
        let enrol = Order::Product {
            kind: "downstream_enrol".into(),
            body: ProductOrder {
                params: params.clone(),
                fetch: None,
            },
        };
        let a = execute(&b, &ctx, 7, &enrol, &mut NoAsk);
        assert!(a.ok && a.detail.contains("downstream_enrol"), "{a:?}");
        let put = Order::Product {
            kind: "bundle_put".into(),
            body: ProductOrder {
                params,
                fetch: Some(Fetch {
                    id: "ab".repeat(32),
                    bytes: 3,
                    param: "bundle".into(),
                }),
            },
        };
        assert!(execute(&b, &ctx, 8, &put, &mut Shop(Ok(b"abc".to_vec()))).ok);
        let failed = execute(&b, &ctx, 9, &put, &mut Shop(Err("gone".into())));
        assert!(
            !failed.ok && failed.detail.contains("was not pulled"),
            "{failed:?}"
        );
        assert_eq!(
            b.did.borrow().len(),
            2,
            "a file not pulled runs nothing: {:?}",
            b.did.borrow()
        );
        assert!(b.did.borrow()[1].contains("Some(3)"));
        assert!(
            b.did.borrow()[1].ends_with("by 8"),
            "the box is told the verified order it runs for: {:?}",
            b.did.borrow()
        );
        let twin = Fake {
            twin: true,
            ..Fake::default()
        };
        assert!(
            !execute(&twin, &ctx, 10, &enrol, &mut NoAsk).ok,
            "a twin runs none"
        );
    }

    /// ★ MONETIZE-FIX: Holger's own kinds decode as product orders; a bundle_put with no file,
    /// and a body that is not a product order, are refused by name.
    #[test]
    fn holgers_own_kinds_decode_and_a_bundle_needs_its_file() {
        use crate::order::{decode, holger_kind};
        for k in holger_kind::PRODUCT {
            if k == holger_kind::BUNDLE_PUT {
                continue;
            }
            assert!(
                matches!(
                    decode(1, k, r#"{"params":{"name":"x"}}"#),
                    Ok(Order::Product { .. })
                ),
                "{k}"
            );
        }
        assert!(
            decode(1, holger_kind::BUNDLE_PUT, r#"{"params":{"path":"b"}}"#)
                .unwrap_err()
                .contains("no file")
        );
        assert!(decode(1, holger_kind::DOWNSTREAM_ENROL, r#"[1]"#)
            .unwrap_err()
            .contains("does not decode"));
        assert!(decode(1, "downstream_destroy", "{}")
            .unwrap_err()
            .contains("does not know"));
    }
}