monetize-estate-side 0.3.3

//! **The pull channel as a jera job** (house rule: server background work is a jera
//! ChainQueue job, never a raw thread). One job = one round: report, long-poll the
//! orders (≤ 25 s), verify and run each, and report again at once when anything ran so
//! the acks reach monetize. A small ticker submits the round under ONE idempotency key,
//! so a round is never queued twice; the job's ledger row says what it did.

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock};
use std::time::Duration;

use jera::jobs::{ChainQueue, ChainSpec, JobSink, Priority};
use monetize_embed::fleet::LONG_POLL_MAX_MS;

use super::puller::PumpBox;
use super::puller::{Link, Puller};

/// The ledger kind of one round.
pub const KIND: &str = "fleet.round";
/// The one key every round is submitted under.
pub const KEY: &str = "fleet-round";

/// **The role this box pulls as, asked at every round** (A7 row 1): nordisk-pairing
/// `elect::pull_role` over what the box knows NOW. `None`: not elected yet, the round waits
/// and reports nothing.
pub type RoleNow = Arc<dyn Fn() -> Option<String> + Send + Sync>;

/// Everything one round needs.
pub struct Duty {
    pub puller: Mutex<Puller>,
    pub link: Mutex<Box<dyn Link + Send>>,
    pub boxed: Arc<dyn PumpBox + Send + Sync>,
    /// The key orders verify under: replaced live when a later feed names one
    /// ([`Duty::set_key`]).
    monetize_pubkey: RwLock<Option<[u8; 32]>>,
    /// The role at each round; `None` = the puller's own, fixed (tests, an enrolment).
    role: Option<RoleNow>,
    /// The pause before the next round, as monetize asked (ms).
    pub pause_ms: AtomicU64,
    /// When this channel started: an `unauthenticated` refusal inside
    /// `nordisk_pairing::fleet::NOT_YET_KNOWN_GRACE` of it is "not known yet" ([`after_round`]).
    started: std::time::Instant,
    /// The `serving` and `data` words of the last report monetize TOOK (`None`: none taken
    /// yet): the first one taken, and every change of either word, is said at [`SAID_AT`]
    /// ([`accepted_line`]; lane F12: test3-holger-rs's main said nothing once its reports landed).
    accepted: Mutex<Option<(String, String)>>,
}

impl Duty {
    pub fn new(
        puller: Puller,
        link: Box<dyn Link + Send>,
        boxed: Arc<dyn PumpBox + Send + Sync>,
        monetize_pubkey: Option<[u8; 32]>,
    ) -> Arc<Duty> {
        Duty::with_role(puller, link, boxed, monetize_pubkey, None)
    }

    /// [`Duty::new`] whose role is asked at every round (`role`).
    pub fn with_role(
        puller: Puller,
        link: Box<dyn Link + Send>,
        boxed: Arc<dyn PumpBox + Send + Sync>,
        monetize_pubkey: Option<[u8; 32]>,
        role: Option<RoleNow>,
    ) -> Arc<Duty> {
        Arc::new(Duty {
            puller: Mutex::new(puller),
            link: Mutex::new(link),
            boxed,
            monetize_pubkey: RwLock::new(monetize_pubkey),
            role,
            pause_ms: AtomicU64::new(super::config::DEFAULT_EVERY.as_millis() as u64),
            started: std::time::Instant::now(),
            accepted: Mutex::new(None),
        })
    }

    /// The key orders verify under now.
    pub fn monetize_pubkey(&self) -> Option<[u8; 32]> {
        *self
            .monetize_pubkey
            .read()
            .unwrap_or_else(|e| e.into_inner())
    }

    /// **A key fed after the start**, applied from the next round on (no restart).
    pub fn set_key(&self, key: Option<[u8; 32]>) {
        *self
            .monetize_pubkey
            .write()
            .unwrap_or_else(|e| e.into_inner()) = key;
    }

    /// **A door fed after the start** (another `fleet`, `fleet_at`, `fleet_ca` or estate):
    /// the next round dials `link` as `instance`.
    pub fn relink(&self, link: Box<dyn Link + Send>, instance: &str) {
        *self.link.lock().unwrap_or_else(|e| e.into_inner()) = link;
        self.puller
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .me
            .instance = instance.to_string();
    }

    /// **One round**, and what it did (the job row's detail).
    pub fn round(&self, wait_ms: u64) -> anyhow::Result<serde_json::Value> {
        let mut p = self.puller.lock().unwrap_or_else(|e| e.into_inner());
        if let Some(role) = &self.role {
            match role() {
                None => {
                    return Ok(
                        serde_json::json!({ "reported": false, "waiting": "the installer's door has not elected this box's role yet: nothing is reported under a guess" }),
                    )
                }
                Some(r) if r != p.me.role => {
                    log::warn!(
                        "fleet: this box now reports as `{r}` (it reported as `{}`)",
                        p.me.role
                    );
                    p.me.role = r;
                }
                Some(_) => {}
            }
        }
        let mut link = self.link.lock().unwrap_or_else(|e| e.into_inner());
        let b = self.boxed.as_ref();
        let now = crate::now_unix_ms;
        let key = self.monetize_pubkey();
        let health = b.health();
        let words = (
            health.get("serving").cloned().unwrap_or_default(),
            health.get("data").cloned().unwrap_or_default(),
        );
        let reported = p.report(
            link.as_mut(),
            health,
            b.product_report(),
            b.own_review(),
            now(),
        );
        if reported.is_ok() {
            let mut last = self.accepted.lock().unwrap_or_else(|e| e.into_inner());
            if let Some(line) = accepted_line(last.as_ref(), &words, &p.me.role) {
                log::log!(SAID_AT, "{line}");
            }
            *last = Some(words);
        }
        if let Ok(next) = &reported {
            if *next > 0 {
                self.pause_ms.store(
                    (*next).clamp(1_000, super::config::DEFAULT_EVERY.as_millis() as u64),
                    Ordering::Relaxed,
                );
            }
        }
        let pulled = p.pull(
            link.as_mut(),
            b,
            key.as_ref(),
            wait_ms.min(LONG_POLL_MAX_MS),
            now(),
        );
        let ran = pulled.as_ref().map(|x| !x.acks.is_empty()).unwrap_or(false);
        // The acks ride a report at once: the guardian waits on them. One that does not land
        // is said (the acks stay pending and ride the next round's report).
        let acks_reported = ran.then(|| {
            p.report(
                link.as_mut(),
                b.health(),
                b.product_report(),
                b.own_review(),
                now(),
            )
        });
        let acks_report_error = acks_reported
            .as_ref()
            .and_then(|r| r.as_ref().err())
            .map(|f| f.to_string());
        if let Some(e) = &acks_report_error {
            log::log!(SAID_AT, "fleet: the report carrying this round's acks FAILED: {e} — they ride the next round's report");
        }
        if let Ok(x) = &pulled {
            for a in &x.acks {
                if a.ok {
                    log::info!("fleet: monetize's order {} ran: {}", a.seq, a.detail);
                } else {
                    log::warn!(
                        "fleet: monetize's order {} was refused: {}",
                        a.seq,
                        a.detail
                    );
                }
            }
            for f in &x.forged {
                log::error!("fleet: an order that is NOT monetize's was offered and ignored: {f}");
            }
        }
        if let Some(line) = report_round_line(&reported) {
            // A7 row 31 (Gunnar 59fa3800): "not known yet" is said at INFO and asked again
            // within 10 s; past the grace, or any other refusal, at WARN.
            let pause = Duration::from_millis(self.pause_ms.load(Ordering::Relaxed));
            let (level, pause) = after_round(&reported, self.started.elapsed(), pause);
            self.pause_ms
                .store(pause.as_millis() as u64, Ordering::Relaxed);
            if level == log::Level::Info {
                log::info!("{line} (monetize does not know this box yet: the estate's owner opens its window as the install feeds it; asked again in {} s)", pause.as_secs());
            } else {
                log::log!(level, "{line}");
            }
        }
        let report_err = reported.as_ref().err().map(|f| f.to_string());
        let detail = serde_json::json!({
            "reported": reported.is_ok(),
            "report_error": report_err,
            "acks_report_error": acks_report_error,
            "acks": pulled.as_ref().map(|x| x.acks.iter().map(|a| serde_json::json!({"seq": a.seq, "ok": a.ok, "detail": a.detail})).collect::<Vec<_>>()).unwrap_or_default(),
            "forged": pulled.as_ref().map(|x| x.forged.len()).unwrap_or(0),
            "pull_error": pulled.as_ref().err().map(|f| f.to_string()),
        });
        match (&reported, &pulled) {
            (Err(r), Err(o)) if o.code != "no_key" => {
                Err(anyhow::anyhow!("report: {r}; orders: {o}"))
            }
            _ => Ok(detail),
        }
    }
}

/// **The level the pull channel's own story is said at** (lane R10): an estate box logs at
/// `warn` by default (holger-server-lib `DEFAULT_LOG_FILTER`), so the channel's start, every
/// refused or failed report round and the boot log's `STAGE report` are said at WARN —
/// at `info` none of them was ever seen on a box (test3-holger-rs, 2026-10-20; Holger's lane R10).
pub const SAID_AT: log::Level = log::Level::Warn;

/// Pure: **the level a refused round is said at, and the pause before the next** (A7 row 31,
/// Gunnar 59fa3800): nordisk-pairing `fleet::after_refusal` over the refusal's code and the
/// time since the channel started. `unauthenticated` inside the grace (monetize does not know
/// this box YET: the owner's pin window opens as the install feeds it) is said at INFO and
/// asked again within 10 s; anything else at [`SAID_AT`], after `pause`.
pub fn after_round(
    reported: &Result<u64, super::puller::Fault>,
    since_start: Duration,
    pause: Duration,
) -> (log::Level, Duration) {
    match reported {
        Err(f) => match nordisk_pairing::fleet::after_refusal(&f.code, since_start, pause) {
            (p, true) => (log::Level::Info, p),
            (p, false) => (SAID_AT, p),
        },
        Ok(_) => (SAID_AT, pause),
    }
}

/// Pure: **what a report round says when it did not land** — monetize's refusal in its own
/// words (`refused`, `unauthenticated`, `malformed`), or why the call failed; `None` when
/// the report was taken.
pub fn report_round_line(reported: &Result<u64, super::puller::Fault>) -> Option<String> {
    let f = reported.as_ref().err()?;
    Some(match f.code.as_str() {
        "refused" | "unauthenticated" | "malformed" => format!(
            "fleet: monetize REFUSED this round's report ({}): {}",
            f.code, f.message
        ),
        _ => format!(
            "fleet: this round's report FAILED ({}): {} — the next round tries again",
            f.code, f.message
        ),
    })
}

/// Pure: **what a report monetize TOOK says** (lane F12): the first one taken, and any
/// later one whose `serving` or `data` word differs from the last one taken; `None` for a
/// report that says what the last one said (a quiet channel is a working one). `words` is
/// (`serving`, `data`); an absent word is said as `nothing`.
pub fn accepted_line(
    last: Option<&(String, String)>,
    words: &(String, String),
    role: &str,
) -> Option<String> {
    let shown = |v: &str| {
        if v.trim().is_empty() {
            "nothing".to_string()
        } else {
            v.trim().to_string()
        }
    };
    let (serving, data) = (shown(&words.0), shown(&words.1));
    match last {
        None => Some(format!("fleet: monetize TOOK this box's first report as `{role}`: serving = {serving}, data = {data}")),
        Some(l) if l != words => Some(format!("fleet: monetize took this box's report as `{role}`: serving = {serving} (was {}), data = {data} (was {})", shown(&l.0), shown(&l.1))),
        Some(_) => None,
    }
}

/// The spec of one round, in `product`'s jera workspace.
pub fn spec(generation: i64, product: &str) -> ChainSpec {
    ChainSpec::new(
        KIND,
        KEY,
        "fleet",
        generation,
        product,
        serde_json::json!({}),
    )
    .with_target("the ONE monetize's pull channel")
    .with_priority(Priority::Normal)
}

/// **Start the pull channel** on `rt`: a ChainQueue of its own (a held long-poll never
/// waits behind other work, nor other work behind it) and the ticker that submits rounds.
pub fn start(rt: &tokio::runtime::Handle, sink: JobSink, duty: Arc<Duty>) -> Arc<ChainQueue> {
    let q = ChainQueue::new(sink);
    let d = Arc::clone(&duty);
    // A32-14, stays a task: the pull channel's jera worker.
    rt.spawn(Arc::clone(&q).run(move |_spec| d.round(LONG_POLL_MAX_MS)));
    let tick = Arc::clone(&q);
    let product = duty
        .puller
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .me
        .product
        .clone();
    // A32-14, stays a task: the pull channel's ticker (it submits the jera rounds).
    rt.spawn(async move {
        let mut generation = 0i64;
        loop {
            generation += 1;
            let _ = tick.submit(spec(generation, &product));
            let pause = duty.pause_ms.load(Ordering::Relaxed).min(5_000);
            tokio::time::sleep(Duration::from_millis(pause)).await;
        }
    });
    q
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::pump::key::BoxKey;
    use crate::pump::puller::{Addressee, Asker, Ctx, PumpBox};
    use crate::pump::puller::{Fault, Signed};
    use monetize_embed::fleet::SignedOrder;

    /// A link that records the role of every call and answers nothing.
    struct Heard(Arc<Mutex<Vec<String>>>);
    impl Link for Heard {
        fn report(&mut self, s: Signed) -> Result<u64, crate::pump::puller::Fault> {
            self.0
                .lock()
                .unwrap()
                .push(format!("report {}/{}", s.call.instance, s.call.role));
            Ok(1_000)
        }
        fn orders(&mut self, s: Signed) -> Result<Vec<SignedOrder>, Fault> {
            self.0
                .lock()
                .unwrap()
                .push(format!("orders {}/{}", s.call.instance, s.call.role));
            Ok(Vec::new())
        }
        fn ask(&mut self, _: Signed) -> Result<crate::pump::puller::Asked, Fault> {
            Err(Fault::new("none", "no asks here"))
        }
    }

    struct Bare;
    impl PumpBox for Bare {
        fn run(
            &self,
            _: &Ctx,
            o: &SignedOrder,
            _: &mut dyn Asker,
        ) -> monetize_embed::fleet::OrderAck {
            monetize_embed::fleet::OrderAck {
                seq: o.seq,
                ok: false,
                detail: "a bare box runs nothing".into(),
                receipt_json: String::new(),
            }
        }
    }

    fn duty(role: Option<RoleNow>, heard: &Arc<Mutex<Vec<String>>>) -> Arc<Duty> {
        let me = Addressee {
            product: "a-product".into(),
            instance: "acme".into(),
            role: "main".into(),
        };
        let p = Puller::new(me, BoxKey::from_seed([3; 32]), "test");
        Duty::with_role(
            p,
            Box::new(Heard(Arc::clone(heard))),
            Arc::new(Bare),
            Some([9; 32]),
            role,
        )
    }

    /// ★ A7 row 1: the role is asked at every round. A box the door has not elected
    /// reports NOTHING; elected twin after its start, it reports as `twin`, never as `main`.
    #[test]
    fn a_box_elected_twin_after_its_start_reports_as_the_twin() {
        let heard = Arc::new(Mutex::new(Vec::new()));
        let elected: Arc<Mutex<Option<Option<&'static str>>>> = Arc::new(Mutex::new(Some(None)));
        let e = Arc::clone(&elected);
        let role: RoleNow =
            Arc::new(move || nordisk_pairing::elect::pull_role("", false, *e.lock().unwrap()));
        let d = duty(Some(role), &heard);
        d.round(0).unwrap();
        assert!(
            heard.lock().unwrap().is_empty(),
            "not elected yet: nothing reported under a guess, {:?}",
            heard.lock().unwrap()
        );
        *elected.lock().unwrap() = Some(Some("twin"));
        d.round(0).unwrap();
        assert_eq!(
            *heard.lock().unwrap(),
            vec!["report acme/twin", "orders acme/twin"]
        );
    }

    /// ★ P4 (lane mon-shared) RED->GREEN: the report that carries a run order's ack back is
    /// never dropped. Its failure is said, and the round's row names it; the ack stays
    /// pending and rides the next round's report.
    #[test]
    fn a_failed_acks_report_is_said_and_the_ack_rides_the_next_round() {
        struct AcksLost(u32);
        impl Link for AcksLost {
            fn report(&mut self, _: Signed) -> Result<u64, Fault> {
                self.0 += 1;
                if self.0 == 2 {
                    Err(Fault::new("unavailable", "the door went away"))
                } else {
                    Ok(1_000)
                }
            }
            fn orders(&mut self, _: Signed) -> Result<Vec<SignedOrder>, Fault> {
                Ok(vec![crate::pump::puller::tests::signed(
                    4,
                    "main",
                    "note",
                    "{}",
                    crate::pump::puller::tests::MONETIZE_SEED,
                )])
            }
            fn ask(&mut self, _: Signed) -> Result<crate::pump::puller::Asked, Fault> {
                Err(Fault::new("none", "no asks here"))
            }
        }
        let me = Addressee {
            product: "a-product".into(),
            instance: "acme".into(),
            role: "main".into(),
        };
        let p = Puller::new(me, BoxKey::from_seed([3; 32]), "test");
        let key = crate::pump::puller::tests::monetize_key().to_bytes();
        let d = Duty::with_role(p, Box::new(AcksLost(0)), Arc::new(Bare), Some(key), None);
        let row = d.round(0).unwrap();
        assert_eq!(row["acks"].as_array().map(|a| a.len()), Some(1), "{row}");
        let said = row["acks_report_error"].as_str().unwrap_or_default();
        assert!(
            said.contains("the door went away"),
            "the failed acks report is in the row: {row}"
        );
        assert_eq!(
            d.puller.lock().unwrap().pulled.pending.len(),
            1,
            "the ack stays pending for the next report"
        );
    }

    /// A key and a door fed after the start reach the next round (no restart).
    #[test]
    fn a_late_key_and_a_new_door_reach_the_next_round() {
        let heard = Arc::new(Mutex::new(Vec::new()));
        let d = duty(None, &heard);
        d.set_key(Some([1; 32]));
        assert_eq!(d.monetize_pubkey(), Some([1; 32]));
        let moved = Arc::new(Mutex::new(Vec::new()));
        d.relink(Box::new(Heard(Arc::clone(&moved))), "acme2");
        d.round(0).unwrap();
        assert!(heard.lock().unwrap().is_empty());
        assert_eq!(
            *moved.lock().unwrap(),
            vec!["report acme2/main", "orders acme2/main"]
        );
    }

    /// ★ R10 RED→GREEN: a box logs at `warn` by default, so the pull channel's story is said
    /// at WARN or it is never seen.
    #[test]
    fn the_pull_channels_story_is_said_at_warn() {
        assert!(
            SAID_AT <= log::Level::Warn,
            "{SAID_AT:?} is dropped by a box's default `warn` filter"
        );
    }

    /// ★ F12 (test3-holger-rs 2026-10-21: after its pin its main said nothing more): the first
    /// report monetize TOOK is said, and so is every change of its `serving` or `data` word; a
    /// report that says what the last one said is quiet. The duty keeps the words it last had
    /// taken, and a refused round keeps none.
    #[test]
    fn the_first_report_taken_and_every_change_of_its_words_is_said() {
        let w = |s: &str, d: &str| (s.to_string(), d.to_string());
        let first =
            accepted_line(None, &w("yes", "ok"), "main").expect("the first report taken is said");
        assert!(
            first.contains("TOOK")
                && first.contains("`main`")
                && first.contains("serving = yes")
                && first.contains("data = ok"),
            "{first}"
        );
        assert_eq!(
            accepted_line(Some(&w("yes", "ok")), &w("yes", "ok"), "main"),
            None,
            "the same words again are quiet"
        );
        let changed = accepted_line(Some(&w("yes", "ok")), &w("no", "fresh"), "main")
            .expect("a change is said");
        assert!(
            changed.contains("serving = no (was yes)") && changed.contains("data = fresh (was ok)"),
            "{changed}"
        );
        assert!(accepted_line(None, &w("", ""), "twin")
            .unwrap()
            .contains("serving = nothing, data = nothing"));

        let heard = Arc::new(Mutex::new(Vec::new()));
        let d = duty(None, &heard);
        assert!(d.accepted.lock().unwrap().is_none());
        d.round(0).unwrap();
        assert!(
            d.accepted.lock().unwrap().is_some(),
            "an accepted round keeps the words it said"
        );
    }

    /// ★ A7 row 31 (Gunnar 59fa3800): `unauthenticated` inside the grace is "not known yet":
    /// said at INFO, asked again within 10 s. Past the grace, or any other refusal: WARN and
    /// the pause as before. The duty's own round keeps the pause it chose.
    #[test]
    fn a_box_not_known_yet_is_said_quietly_and_asked_again_soon_inside_the_grace() {
        let every = Duration::from_secs(30);
        let unauth: Result<u64, Fault> = Err(Fault::new(
            "unauthenticated",
            "monetize does not know this box key",
        ));
        assert_eq!(
            after_round(&unauth, Duration::from_secs(60), every),
            (log::Level::Info, Duration::from_secs(10))
        );
        assert_eq!(
            after_round(
                &unauth,
                nordisk_pairing::fleet::NOT_YET_KNOWN_GRACE + Duration::from_secs(1),
                every
            ),
            (SAID_AT, every),
            "past the grace: WARN"
        );
        assert_eq!(
            after_round(
                &Err(Fault::new("unavailable", "down")),
                Duration::from_secs(60),
                every
            ),
            (SAID_AT, every)
        );
        assert_eq!(
            after_round(&Ok(30_000), Duration::from_secs(60), every).1,
            every,
            "a taken report keeps the pause"
        );
        // A duty refused `unauthenticated` at its first round pauses at most 10 s.
        struct Refused;
        impl Link for Refused {
            fn report(&mut self, _: Signed) -> Result<u64, Fault> {
                Err(Fault::new("unauthenticated", "not known"))
            }
            fn orders(&mut self, _: Signed) -> Result<Vec<SignedOrder>, Fault> {
                Err(Fault::new("unauthenticated", "not known"))
            }
            fn ask(&mut self, _: Signed) -> Result<crate::pump::puller::Asked, Fault> {
                Err(Fault::new("none", "no asks here"))
            }
        }
        let me = Addressee {
            product: "a-product".into(),
            instance: "acme".into(),
            role: "main".into(),
        };
        let d = Duty::new(
            Puller::new(me, BoxKey::from_seed([3; 32]), "test"),
            Box::new(Refused),
            Arc::new(Bare),
            Some([9; 32]),
        );
        let _ = d.round(0);
        assert!(
            d.pause_ms.load(Ordering::Relaxed) <= 10_000,
            "{}",
            d.pause_ms.load(Ordering::Relaxed)
        );
    }

    /// ★ R10 RED→GREEN: every refused or failed report round is said, with monetize's words.
    #[test]
    fn a_refused_or_failed_report_round_is_said_with_monetizes_words() {
        let refused = report_round_line(&Err(Fault::new(
            "refused",
            "no estate acme-7 is sold to this box key",
        )))
        .unwrap();
        assert!(
            refused.starts_with("fleet: ")
                && refused.contains("REFUSED")
                && refused.contains("no estate acme-7 is sold to this box key"),
            "{refused}"
        );
        let failed = report_round_line(&Err(Fault::new(
            "unavailable",
            "tcp connect error: Connection refused",
        )))
        .unwrap();
        assert!(
            failed.contains("FAILED")
                && failed.contains("unavailable")
                && failed.contains("tcp connect error"),
            "{failed}"
        );
        assert_eq!(
            report_round_line(&Ok(30_000)),
            None,
            "a taken report says nothing"
        );
    }

    /// ★ P4a: a round is a jera job in the PRODUCT's workspace (it was always "holger"), under
    /// the one idempotency key.
    #[test]
    fn a_round_is_one_job_in_the_products_workspace() {
        let s = spec(7, "a-product");
        assert_eq!(
            (
                s.kind.as_str(),
                s.key.as_str(),
                s.group.as_str(),
                s.generation,
                s.workspace.as_str()
            ),
            (KIND, KEY, "fleet", 7, "a-product")
        );
        assert_eq!(s.target, "the ONE monetize's pull channel");
    }
}