holger-fleet 2.0.0

//! **Holger's box on the shared pump** (P4a): the pump is monetize-estate-side's
//! ([`monetize_estate_side::pump`]); what is Holger's is its executor ([`crate::exec`]) over its
//! own order kinds ([`crate::order`]). [`Pumped`] is the one seam between the two.

use std::collections::BTreeMap;

use monetize_embed::fleet::{OrderAck, SignedOrder};
use monetize_estate_side::pump::{grow_set, Asker, Ctx, PumpBox};

use crate::exec::{execute, GrowSet, HolgerBox};
use crate::order::decode;

/// **A Holger box as the shared pump runs it**: its health words, its product report, its own
/// twin review, and every order decoded by Holger's kinds and run through Holger's verbs. Holds
/// any handle to a [`HolgerBox`] (`&b`, `Arc<dyn HolgerBox + Send + Sync>`).
pub struct Pumped<B>(pub B);

impl<B, T> PumpBox for Pumped<B>
where
    B: std::ops::Deref<Target = T>,
    T: HolgerBox + ?Sized,
{
    fn health(&self) -> BTreeMap<String, String> {
        self.0.health()
    }
    fn product_report(&self) -> Option<String> {
        self.0.product_report()
    }
    fn own_review(&self) -> Option<String> {
        self.0.own_review()
    }
    fn run(&self, ctx: &Ctx, o: &SignedOrder, ask: &mut dyn Asker) -> OrderAck {
        match decode(o.seq, &o.kind, &o.body) {
            Ok(order) => execute(&*self.0, ctx, o.seq, &order, &mut GrowVia(ask)),
            Err(detail) => OrderAck { seq: o.seq, ok: false, detail, receipt_json: String::new() },
        }
    }
}

/// A pulled growth's `Fleet.Ask{grow_set}` on the order's own link (the shared pump's).
struct GrowVia<'a>(&'a mut dyn Asker);

impl GrowSet for GrowVia<'_> {
    fn grow_set(&mut self, gib: u64) -> Result<String, String> {
        grow_set(self.0, gib)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::exec::tests::{fact, Fake};
    use crate::order::tests::{me, monetize_key, signed, MONETIZE_SEED};
    use monetize_embed::fleet::{Ask, OrdersAfter, Report, SignedOrder};
    use monetize_estate_side::pump::{Asked, BoxKey, Fault, Link, Puller, Signed};
    use std::collections::BTreeMap;

    /// Monetize in memory: a queue, and every call it heard.
    #[derive(Default)]
    struct Guardian {
        queue: Vec<SignedOrder>,
        heard: Vec<Signed>,
    }

    impl Link for Guardian {
        fn report(&mut self, s: Signed) -> Result<u64, Fault> {
            let r: Report = serde_json::from_str(&s.call.body).unwrap();
            self.queue.retain(|o| !r.acked.iter().any(|a| a.seq == o.seq));
            self.heard.push(s);
            Ok(30_000)
        }
        fn orders(&mut self, s: Signed) -> Result<Vec<SignedOrder>, Fault> {
            let after: OrdersAfter = serde_json::from_str(&s.call.body).unwrap();
            self.heard.push(s);
            Ok(self.queue.iter().filter(|o| o.seq > after.after).cloned().collect())
        }
        fn ask(&mut self, s: Signed) -> Result<Asked, Fault> {
            let _: Ask = serde_json::from_str(&s.call.body).unwrap();
            self.heard.push(s);
            Ok(Asked { case_id: "c-1".into(), words: "resized".into(), url: String::new() })
        }
    }

    fn mkey() -> [u8; 32] {
        monetize_key().to_bytes()
    }

    /// ★ P4a: Holger's executor over the SHARED pump (monetize-estate-side): a fact is pulled,
    /// run through Holger's verbs and acked, once.
    #[test]
    fn an_entitlement_is_pulled_run_and_acked_in_the_next_report_and_never_twice() {
        let b = Fake::default();
        let mut g = Guardian::default();
        g.queue.push(signed(4, "main", "entitlement", &serde_json::to_string(&fact("libs", 5, MONETIZE_SEED)).unwrap(), MONETIZE_SEED));
        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
        let pulled = p.pull(&mut g, &Pumped(&b), Some(&mkey()), 0, 10_000).unwrap();
        assert_eq!(pulled.acks.len(), 1);
        assert!(pulled.acks[0].ok, "{pulled:?}");
        assert!(b.facts.borrow().contains_key("libs"));
        p.report(&mut g, BTreeMap::from([("serving".into(), "yes".into())]), Some("{}".into()), None, 10_001).unwrap();
        let r: Report = serde_json::from_str(&g.heard.last().unwrap().call.body).unwrap();
        assert_eq!(r.acked.iter().map(|a| (a.seq, a.ok)).collect::<Vec<_>>(), vec![(4, true)]);
        assert_eq!(r.product.as_deref(), Some("{}"));
        assert!(g.queue.is_empty() && p.pulled.pending.is_empty());
        assert!(p.pull(&mut g, &Pumped(&b), Some(&mkey()), 0, 10_002).unwrap().acks.is_empty(), "nothing runs twice");
    }

    #[test]
    fn a_forged_order_never_runs_never_acks_and_never_moves_the_sequence() {
        let b = Fake::default();
        let mut g = Guardian::default();
        g.queue.push(signed(u64::MAX, "main", "purge_tenant", r#"{"tenant":"libs"}"#, [9u8; 32]));
        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
        let pulled = p.pull(&mut g, &Pumped(&b), Some(&mkey()), 0, 10_000).unwrap();
        assert!(pulled.acks.is_empty() && pulled.forged.len() == 1);
        assert_eq!(p.pulled.last_seq, 0);
        assert!(b.did.borrow().is_empty());
    }

    #[test]
    fn without_a_monetize_key_nothing_is_pulled_and_monetize_is_not_even_asked() {
        let mut g = Guardian::default();
        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
        assert_eq!(p.pull(&mut g, &Pumped(&Fake::default()), None, 0, 10_000).unwrap_err().code, "no_key");
        assert!(g.heard.is_empty());
    }

    #[test]
    fn the_first_call_carries_the_box_key_and_later_ones_do_not() {
        let mut g = Guardian::default();
        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
        p.report(&mut g, BTreeMap::new(), None, None, 1).unwrap();
        p.report(&mut g, BTreeMap::new(), None, None, 2).unwrap();
        assert_eq!(g.heard[0].box_key, p.box_public().to_vec());
        assert!(g.heard[1].box_key.is_empty());
    }

    #[test]
    fn a_twins_own_review_rides_every_report_and_an_ordered_one_the_next_only() {
        let mut g = Guardian::default();
        let mut p = Puller::new(me("twin"), BoxKey::from_seed([4u8; 32]), "test");
        let b = Fake { twin: true, ..Default::default() };
        g.queue.push(signed(1, "twin", "review_twin", "{}", MONETIZE_SEED));
        p.pull(&mut g, &Pumped(&b), Some(&mkey()), 0, 10).unwrap();
        p.report(&mut g, BTreeMap::new(), None, None, 11).unwrap();
        p.report(&mut g, BTreeMap::new(), None, None, 12).unwrap();
        p.report(&mut g, BTreeMap::new(), None, Some(r#"{"own":1}"#.into()), 13).unwrap();
        let said: Vec<Option<String>> = g.heard.iter().filter(|s| s.call.call == "report").map(|s| serde_json::from_str::<Report>(&s.call.body).unwrap().twin_review).collect();
        assert!(said[0].as_deref().is_some_and(|r| r.contains("complete")), "{said:?}");
        assert_eq!(said[1], None);
        assert_eq!(said[2].as_deref(), Some(r#"{"own":1}"#));
    }

    #[test]
    fn the_puller_remembers_what_it_took_across_a_restart() {
        let dir = tempfile::tempdir().unwrap();
        let mut g = Guardian::default();
        g.queue.push(signed(6, "main", "grow_room", r#"{"need_bytes":5,"reason":"r"}"#, MONETIZE_SEED));
        let mut p = Puller::at(dir.path(), me("main"), "test").unwrap();
        let key = p.box_public();
        p.pull(&mut g, &Pumped(&Fake::default()), Some(&mkey()), 0, 10_000).unwrap();
        drop(p);
        let p = Puller::at(dir.path(), me("main"), "test").unwrap();
        assert_eq!((p.box_public(), p.pulled.last_seq, p.pulled.pending.len()), (key, 6, 1));
    }
}