monetize-embed 0.3.1

//! **Test support** (monetize v2.0 P1, A24 §7 step 0): [`PullingBox`], the box on the pull
//! channel modelled, moved here from `monetize-product-gunnar::testing::pulling` so the server's
//! tests keep it when the native product crates go (P2). Behind the `testing` feature; never in
//! a shipped binary.
//!
//! **A gunnar box on the pull channel, modelled** (5.4): what an estate's main or twin does
//! with monetize — report its health, pull the signed orders queued for it, VERIFY each under
//! `monetize_pubkey` (`crate::fleet::verify_order`, the same form gunnar-repair checks)
//! and run the verb, then ack it in its next report. Transport-agnostic: it makes the signed
//! calls and takes the orders; the test carries them (in process, or over gRPC).
//!
//! The verbs are MODELLED here on the box's health words, the way the real ones change them —
//! the real verbs are gunnar-repair's (gunnar `crates/gunnar-repair`), run by the estate.

use std::collections::BTreeMap;

use ed25519_dalek::{Signer as _, SigningKey, VerifyingKey};
use crate::fleet::{ask, call, kind, verify_order, Ask, FleetCall, GrowGoAhead, GrowSetAsk, OrderAck, OrdersAfter, Report, SignedOrder};
use crate::{EntitlementFact, Snapshot};

/// The product id a gunnar box calls home as.
pub const PRODUCT: &str = "gunnar";

/// One box: its key, what it says of itself, and what it ran.
pub struct PullingBox {
    pub instance: String,
    pub role: String,
    key: SigningKey,
    monetize: VerifyingKey,
    /// The box's own health words (`serving`, `data`, `serves_at`, `twin`, `primary`, `epoch`, …).
    pub health: BTreeMap<String, String>,
    /// The highest order seq it executed (replay protection).
    pub last_seq: u64,
    acks: Vec<OrderAck>,
    /// Every order it ran (`kind`), and every one it refused (`refused:<why>`).
    pub ran: Vec<String>,
    /// A verb the box runs and FAILS (models a broken twin, say).
    pub failing: Option<String>,
    /// The entitlement facts it holds, by tenant (what `restate_entitlements` applied).
    pub entitled: BTreeMap<String, EntitlementFact>,
    /// The `issued_unix_ms` of every restatement snapshot it verified and applied.
    pub restated: Vec<u64>,
    /// The paid data step it verified (`grow_go_ahead`): GB a disk its own data may grow to.
    pub grow_to: Option<u64>,
}

impl PullingBox {
    /// A box of `instance` in `role` with its fleet key from `seed`, trusting `monetize_pubkey`.
    pub fn new(instance: &str, role: &str, seed: [u8; 32], monetize_pubkey: [u8; 32], serves_at: &str) -> Self {
        let mut health = BTreeMap::new();
        health.insert("serving".into(), if role == "main" { "yes" } else { "no" }.into());
        health.insert("data".into(), "ok".into());
        health.insert("serves_at".into(), serves_at.into());
        health.insert(if role == "main" { "twin" } else { "primary" }.into(), if role == "main" { "paired" } else { "live" }.into());
        Self {
            instance: instance.into(),
            role: role.into(),
            key: SigningKey::from_bytes(&seed),
            monetize: VerifyingKey::from_bytes(&monetize_pubkey).expect("monetize's public key"),
            health,
            last_seq: 0,
            acks: Vec::new(),
            ran: Vec::new(),
            failing: None,
            entitled: BTreeMap::new(),
            restated: Vec::new(),
            grow_to: None,
        }
    }

    /// The box's fleet public key (what the installer names, or what it pins on first call).
    pub fn public_key(&self) -> [u8; 32] {
        self.key.verifying_key().to_bytes()
    }

    pub fn set(&mut self, word: &str, value: &str) -> &mut Self {
        self.health.insert(word.into(), value.into());
        self
    }

    pub fn word(&self, word: &str) -> &str {
        self.health.get(word).map(String::as_str).unwrap_or("")
    }

    /// One signed call: the call and its signature.
    pub fn call(&self, which: &str, body: String, now_unix_ms: u64) -> (FleetCall, Vec<u8>) {
        let c = FleetCall { product: PRODUCT.into(), instance: self.instance.clone(), role: self.role.clone(), call: which.into(), body, at_unix_ms: now_unix_ms };
        let sig = self.key.sign(&crate::fleet::fleet_call_message(&c)).to_bytes().to_vec();
        (c, sig)
    }

    /// `Fleet.Report`: its health and the acks of the orders it ran since the last one (taken).
    pub fn report(&mut self, now_unix_ms: u64) -> (FleetCall, Vec<u8>) {
        let r = Report { build: "gunnar-mock".into(), health: self.health.clone(), acked: std::mem::take(&mut self.acks), twin_review: None, product: None, measured_unix_ms: now_unix_ms };
        self.call(call::REPORT, serde_json::to_string(&r).expect("a report"), now_unix_ms)
    }

    /// The acks a report that was refused must carry again.
    pub fn unreport(&mut self, report: &FleetCall) {
        if let Ok(r) = serde_json::from_str::<Report>(&report.body) {
            let mut back = r.acked;
            back.append(&mut self.acks);
            self.acks = back;
        }
    }

    /// `Fleet.Orders`: the orders after the last one it ran, held up to `wait_ms`.
    pub fn orders(&self, wait_ms: u64, now_unix_ms: u64) -> (FleetCall, Vec<u8>) {
        self.call(call::ORDERS, serde_json::to_string(&OrdersAfter { after: self.last_seq, wait_ms }).expect("orders"), now_unix_ms)
    }

    /// `Fleet.Ask`.
    pub fn ask(&self, remedy: &str, now_unix_ms: u64) -> (FleetCall, Vec<u8>) {
        self.call(call::ASK, serde_json::to_string(&Ask { kind: remedy.into(), body: serde_json::json!({"why": "the box asked"}) }).expect("ask"), now_unix_ms)
    }

    /// `Fleet.Ask{grow_set}`: its own data disks sized at monetize's provider to `gib` a disk
    /// (0 = what it is paid to; no `members` = all of its own).
    pub fn ask_grow_set(&self, members: &[String], gib: u64, now_unix_ms: u64) -> (FleetCall, Vec<u8>) {
        let body = serde_json::to_value(GrowSetAsk { members: members.to_vec(), gib }).expect("a grow_set ask");
        self.call(call::ASK, serde_json::to_string(&Ask { kind: ask::GROW_SET.into(), body }).expect("ask"), now_unix_ms)
    }

    /// **Run the orders it pulled**: each verified under monetize's key (and for this box,
    /// newer than the last one), then the verb; the ack waits for the next report. Answers the
    /// kinds it ran.
    pub fn run(&mut self, orders: &[SignedOrder], now_unix_ms: u64) -> Vec<String> {
        let mut ran = Vec::new();
        for o in orders {
            if let Err(e) = verify_order(o, &self.monetize, (PRODUCT, &self.instance, &self.role), now_unix_ms, self.last_seq) {
                self.ran.push(format!("refused:{e}"));
                continue;
            }
            self.last_seq = o.seq;
            let (ok, detail) = self.verb(o, now_unix_ms);
            self.ran.push(o.kind.clone());
            ran.push(o.kind.clone());
            let receipt = serde_json::json!({"verb": o.kind, "ok": ok, "detail": detail, "at": now_unix_ms}).to_string();
            self.acks.push(OrderAck { seq: o.seq, ok, detail, receipt_json: receipt });
        }
        ran
    }

    /// The snapshot verified under monetize's key (envelope and every fact), applied newer-wins
    /// per tenant, nothing deleted — and only by a box that serves.
    fn restate(&mut self, body: &serde_json::Value) -> (bool, String) {
        if self.word("serving") != "yes" {
            return (false, format!("the {} does not serve (serving={}): no entitlements applied", self.role, self.word("serving")));
        }
        let snap: Snapshot = match body.get("snapshot").cloned().map(serde_json::from_value) {
            Some(Ok(s)) => s,
            Some(Err(e)) => return (false, format!("the snapshot does not parse: {e}")),
            None => return (false, "the order carries no snapshot".into()),
        };
        if let Err(e) = crate::signing::verify_snapshot(&snap, &self.monetize) {
            return (false, format!("the snapshot does not verify under monetize's key: {e}"));
        }
        let (mut applied, mut held) = (0, 0);
        for f in snap.facts {
            let newer = self.entitled.get(&f.tenant.0).is_none_or(|h| f.issued_unix_ms.unwrap_or(0) > h.issued_unix_ms.unwrap_or(0));
            if newer {
                self.entitled.insert(f.tenant.0.clone(), f);
                applied += 1;
            } else {
                held += 1;
            }
        }
        self.restated.push(snap.issued_unix_ms);
        (true, format!("restated {applied} entitlement facts ({held} already held as new or newer)"))
    }

    /// The go-ahead verified under monetize's key — for THIS estate, still good — and kept:
    /// what its growth (drain, unmount, `grow_set`, remount) may size its own disks to.
    fn grow_go_ahead(&mut self, body: &str, now_unix_ms: u64) -> (bool, String) {
        let order: GrowGoAhead = match serde_json::from_str(body) {
            Ok(o) => o,
            Err(e) => return (false, format!("the grow_go_ahead does not parse: {e}")),
        };
        let g = match crate::signing::verify_estate_growth(order.go_ahead.as_bytes(), &self.monetize) {
            Ok(g) => g,
            Err(e) => return (false, format!("the go-ahead does not verify under monetize's key: {e}")),
        };
        if g.estate != self.instance {
            return (false, format!("the go-ahead is {}'s, not {}'s", g.estate, self.instance));
        }
        if (g.until_unix_ms as u64) < now_unix_ms {
            return (false, format!("the go-ahead ({}) ran out", g.reference));
        }
        self.grow_to = Some(order.data_gib);
        (true, format!("go-ahead {} verified: the {}'s data may grow to {} GB a disk", g.reference, self.role, order.data_gib))
    }

    fn verb(&mut self, o: &SignedOrder, now_unix_ms: u64) -> (bool, String) {
        if self.failing.as_deref() == Some(o.kind.as_str()) {
            if o.kind == kind::RECHARGE_FROM_TWIN {
                self.set("recharge", "failed");
            }
            return (false, format!("{} failed on the {}", o.kind, self.role));
        }
        let body: serde_json::Value = serde_json::from_str(&o.body).unwrap_or_default();
        let s = |k: &str| body.get(k).map(|v| v.as_str().map(str::to_string).unwrap_or_else(|| v.to_string())).unwrap_or_default();
        match o.kind.as_str() {
            kind::FENCE_PRIMARY => {
                self.set("serving", "fenced").set("epoch", &s("epoch"));
                (true, format!("fenced at epoch {}", s("epoch")))
            }
            kind::RECHARGE_FROM_TWIN => {
                self.set("data", "ok").set("serving", "yes").set("recharge", "done").set("epoch", &s("epoch"));
                (true, format!("recharged from the twin at {}", s("twin")))
            }
            kind::ADOPT_TWIN => {
                self.set("twin", "paired");
                (true, format!("adopted the twin at {} ({})", s("twin"), s("principal")))
            }
            kind::REVIEW_TWIN => (true, "the twin is current".into()),
            kind::RESTATE_ENTITLEMENTS => self.restate(&body),
            kind::GROW_GO_AHEAD => self.grow_go_ahead(&o.body, now_unix_ms),
            other => (false, format!("{other:?} is no verb this box runs")),
        }
    }
}