crates/monetize-ui-server/src/sse.rs
//! **The live feed, over server-sent events — streamed, with polling as the
//! fallback.**
//!
//! monetize now has `Events.Watch`, a server-streaming RPC over a bounded event
//! ring. This process holds ONE such stream and fans every event out to every
//! open browser over SSE. That is the whole point of the lane: a purchase, a
//! ladder move or a ledger row lights the console at once instead of up to five
//! seconds later — and a `Declined` that happened and closed inside one poll
//! tick used to leave no trace on the console at all.
//!
//! # The poller did not go away, and must not
//!
//! A console that goes SILENTLY STALE is the failure mode that matters: it looks
//! live and it is wrong. So the five-second poll is still here, as the fallback,
//! and the switch between the two is announced on the feed itself
//! (`{"kind":"feed","mode":"stream"|"polling"}`) and logged at warn:
//!
//! ```text
//! Events.Watch opens ──▶ mode=stream, the poller PARKS
//! the stream dies ──▶ mode=polling, the poller wakes, one seed poll, then diffs
//! the stream returns ──▶ mode=stream, the poller parks again
//! ```
//!
//! The poller's baseline is reset on the way into polling mode, so the first
//! fallback poll emits nothing: the events it would otherwise "discover" already
//! went out over the stream, and a console that showed each of them twice would
//! be teaching an operator to distrust it.
//!
//! # A gap is passed straight through
//!
//! monetize's ring is bounded and drops the OLDEST events for a subscriber that
//! cannot keep up, telling it so with an `EventGap`. This process does not
//! swallow that: it becomes a `{"kind":"gap"}` line, and every list in the
//! console re-reads on it. A subscriber that silently missed events would be
//! worse than one that polls.
use std::collections::BTreeMap;
use std::convert::Infallible;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use axum::extract::State;
use axum::response::sse::{Event, KeepAlive, Sse};
use futures_util::stream::Stream;
use futures_util::StreamExt;
use monetize_proto as pb;
use monetize_proto::query_ledger_request::By;
use serde_json::{json, Value};
use tokio::sync::{broadcast, Notify};
use crate::api::state_name;
use crate::app::AppState;
use crate::grpc::{Fail, Monetize};
/// The longest this process waits between attempts to re-open `Events.Watch`.
/// Short enough that a monetize restart is picked up inside half a minute,
/// long enough that a monetize which does not serve `Events` at all (an older
/// build) costs a reconnect every 30 s rather than a hot loop.
const MAX_RECONNECT_BACKOFF: Duration = Duration::from_secs(30);
#[derive(Clone)]
pub struct EventHub {
tx: broadcast::Sender<String>,
wake: Arc<Notify>,
/// True while `Events.Watch` is delivering. The poller parks while it is; it
/// is the one bit of state the two tasks share.
streaming: Arc<AtomicBool>,
/// **The latest `feed` line, kept.** A mode change is announced once, on the
/// transition — but a browser that opens the console ten minutes into an
/// outage was not there to hear it, and a console that cannot say whether it
/// is live is the thing this whole lane exists to avoid. So the line is
/// remembered and replayed as the FIRST frame of every SSE connection.
mode: Arc<Mutex<Value>>,
}
/// The `feed` line a hub starts with: nothing has been tried yet, so the honest
/// answer is "not streaming", with the reason spelled out rather than left as a
/// bare mode a reader would have to guess about.
fn initial_mode() -> Value {
json!({ "kind": "feed", "mode": "polling", "message": "the console's server has not opened monetize's event stream yet" })
}
/// What one poll saw, reduced to what a diff needs. Only the FALLBACK uses this.
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct Snapshot {
/// `(product, tenant)` → `order|state|paid_until|source`.
pub facts: BTreeMap<(String, String), String>,
/// open reference → how many rows it has.
pub open: BTreeMap<String, usize>,
}
pub fn fingerprint(f: &pb::EntitlementFact) -> String {
format!("{}|{}|{:?}|{}", f.plan, state_name(f.state()), f.paid_until_unix_ms, f.source)
}
/// The events between two polls. Pure, so it is tested without a server.
pub fn diff(prev: &Snapshot, next: &Snapshot, now_ms: u64) -> Vec<Value> {
let mut out = Vec::new();
for ((product, tenant), fp) in &next.facts {
if prev.facts.get(&(product.clone(), tenant.clone())) != Some(fp) {
let mut parts = fp.splitn(4, '|');
let (order, state) = (parts.next().unwrap_or(""), parts.next().unwrap_or(""));
let source = parts.nth(1).unwrap_or("");
out.push(json!({ "kind": "entitlement", "product": product, "tenant": tenant, "order": order, "state": state, "source": source, "at_unix_ms": now_ms }));
}
}
for (reference, rows) in &next.open {
if prev.open.get(reference) != Some(rows) {
out.push(json!({ "kind": "ledger", "reference": reference, "open": true, "rows": rows, "at_unix_ms": now_ms }));
}
}
for reference in prev.open.keys().filter(|r| !next.open.contains_key(*r)) {
out.push(json!({ "kind": "ledger", "reference": reference, "open": false, "at_unix_ms": now_ms }));
}
out
}
/// **One `monetize.v1.Event` as the console's feed sees it.**
///
/// Pure, and the one translator (one implementation): the SSE handler, the tests and any
/// future consumer read the same shape. `None` for `StreamOpened`, which is a
/// fact about the stream that the stream driver acts on itself — a browser has
/// no use for an epoch it cannot resume from.
pub fn event_to_feed(ev: &pb::Event) -> Option<Value> {
match ev.payload.as_ref()? {
pb::event::Payload::EntitlementChanged(e) => Some(json!({
"kind": "entitlement",
"product": e.product,
"tenant": e.tenant,
// The order the fact came from (the fact's `plan` field on the wire).
"order": e.order,
"state": state_name(e.to_state()),
// Empty when there was no fact before: "a tenant appeared at Paid"
// and "a tenant fell from Paid" are different lines and an empty
// `from_state` is how the console tells them apart.
"from_state": if e.had_previous { state_name(e.from_state()) } else { "" },
"source": e.source,
"pushed": e.pushed,
"at_unix_ms": ev.at_unix_ms,
})),
pb::event::Payload::LedgerAppended(e) => Some(json!({
"kind": "ledger",
"reference": e.reference,
"product": e.product,
"tenant": e.tenant,
"seq": e.seq,
"step": e.step,
"open": e.open,
"at_unix_ms": e.at_unix_ms,
})),
pb::event::Payload::UsageMeasured(e) => Some(json!({
"kind": "usage",
"product": e.product,
"tenant": e.tenant,
"meters": e.meters.iter().map(|m| json!({ "name": m.name, "value": m.value })).collect::<Vec<_>>(),
"at_unix_ms": e.measured_at_unix_ms,
})),
// **A callback monetize refused.** The one line on this feed that a
// stranger can provoke, and the only notice the console gets during an
// outage in which NOTHING else happens: when every delivery is refused
// there is no ledger row, no entitlement move and no usage reading, so
// without this the feed would be silent for exactly as long as the
// failure lasted — which is what happened on 2026-09-14.
//
// `verified` rides along because it says whose words `detail` holds:
// false is a caller who proved nothing, and the console must label it
// rather than present a stranger's sentence as monetize's.
pb::event::Payload::CallbackRefused(r) => Some(json!({
"kind": "callback_refused",
"refusal": r.kind,
"vendor": r.vendor,
"verified": r.verified,
"detail": r.detail,
"body_bytes": r.body_bytes,
"refusal_seq": r.refusal_seq,
"at_unix_ms": r.at_unix_ms,
})),
// **A payment found at the vendor that monetize had no row for.** The
// other line this feed carries during an outage where nothing else
// moves: the refusal above says a delivery was thrown away, this says
// what money was behind it — and, when the poller could place it,
// that it has been handed to the signup engine. The vendor's word,
// over monetize's own key; the amount and the session are what an
// operator acts on.
pb::event::Payload::PaidAtVendor(c) => Some(json!({
"kind": "paid_at_vendor",
"vendor": c.vendor,
"session": c.session,
"reference": c.reference,
"amount_minor": c.amount_minor,
"currency": c.currency.to_ascii_uppercase(),
"state": c.state,
"reason": c.reason,
"paid_at_unix_ms": c.paid_at_unix_ms.max(0),
"at_unix_ms": c.first_seen_unix_ms.max(0),
})),
// **One line of an order's setup** (the test button, 2026-10-11): a
// phase the order moved to, or what the installer said. The console
// re-reads the run it shows; the line rides along for a reader that
// wants it.
pb::event::Payload::SetupLine(l) => Some(json!({
"kind": "setup_line",
"product": l.product,
"tenant": l.tenant,
"order": l.order,
"source": l.source,
"phase": l.phase,
"line": l.line,
"at_unix_ms": l.at_unix_ms,
})),
// Passed through, never swallowed: this is how a console learns that the
// incremental view it is holding has a hole in it.
pb::event::Payload::Gap(g) => Some(json!({
"kind": "gap",
"dropped": g.dropped,
"last_delivered_seq": g.last_delivered_seq,
"resume_seq": g.resume_seq,
"at_unix_ms": ev.at_unix_ms,
})),
pb::event::Payload::StreamOpened(_) => None,
}
}
async fn poll_once(m: &Monetize) -> Result<Snapshot, Fail> {
let mut snap = Snapshot::default();
for p in m.products().list(pb::ListProductsRequest {}).await?.into_inner().products {
for f in m.entitlement().list(pb::ListEntitlementsRequest { product: p.id.clone() }).await?.into_inner().facts {
snap.facts.insert((p.id.clone(), f.tenant.clone()), fingerprint(&f));
}
}
let open = m.ledger().query(pb::QueryLedgerRequest { by: Some(By::Unfinished(true)) }).await?.into_inner();
for r in open.references {
let n = open.rows.iter().filter(|x| x.reference == r).count();
snap.open.insert(r, n);
}
Ok(snap)
}
fn now_ms() -> u64 {
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0)
}
impl EventHub {
/// Start both halves: the `Events.Watch` driver, and the poller that stands
/// in whenever the stream is down. `every` is the fallback poll interval.
pub fn spawn(monetize: Monetize, every: Duration) -> EventHub {
let (tx, _) = broadcast::channel(256);
let hub = EventHub {
tx,
wake: Arc::new(Notify::new()),
streaming: Arc::new(AtomicBool::new(false)),
mode: Arc::new(Mutex::new(initial_mode())),
};
// A32-14, stays a task: the console's event poller, async, for the console's life.
tokio::spawn(poll_forever(monetize.clone(), hub.clone(), every));
// A32-14, stays a task: the console's event stream, async, for the console's life.
tokio::spawn(stream_forever(monetize, hub.clone()));
hub
}
/// A hub with no tasks behind it — for tests that drive [`EventHub::publish`]
/// themselves, and for a `--dry` bring-up.
pub fn detached() -> EventHub {
let (tx, _) = broadcast::channel(256);
EventHub { tx, wake: Arc::new(Notify::new()), streaming: Arc::new(AtomicBool::new(false)), mode: Arc::new(Mutex::new(initial_mode())) }
}
/// Announce which half of the feed is running, and REMEMBER it. Every mode
/// change goes through here (one implementation), so the remembered line and the one on
/// the wire cannot disagree.
fn set_mode(&self, mode: &str, message: &str) {
let line = json!({ "kind": "feed", "mode": mode, "message": message, "at_unix_ms": now_ms() });
*self.mode.lock().unwrap_or_else(|e| e.into_inner()) = line.clone();
self.publish(line);
}
/// The remembered `feed` line, stamped with the time it is being read: what
/// a browser is told the moment it connects.
pub fn mode_line(&self) -> Value {
let mut line = self.mode.lock().unwrap_or_else(|e| e.into_inner()).clone();
line["at_unix_ms"] = json!(now_ms());
line
}
pub fn publish(&self, ev: Value) {
let _ = self.tx.send(ev.to_string());
}
/// Poll now: the console just wrote something. Harmless while streaming —
/// monetize will publish the write's own event either way.
pub fn poke(&self) {
self.wake.notify_one();
}
/// True while the feed is coming off `Events.Watch` rather than the poller.
pub fn is_streaming(&self) -> bool {
self.streaming.load(Ordering::Relaxed)
}
pub fn subscribe(&self) -> broadcast::Receiver<String> {
self.tx.subscribe()
}
}
/// **What the console says when the ONE monetize refuses its key** (stinkies
/// control-refusal-spam): the ONE monetize was not fed this console's
/// `monetize-ui` key (the ONE box laid before the console, or re-laid). No
/// estate runs a monetize of its own, so there is no estate to blame.
pub const UNFED_WORDS: &str = "the ONE monetize was not fed this console's key — re-lay the ONE box so its install feeds monetize_principal.monetize-ui";
/// The wait after the ONE monetize answered `Unauthenticated` `n` times in a row
/// (n ≥ 1): the poll's own interval (`every`), unchanged.
///
/// 30 s, doubling, at most [`UNFED_MAX`] — never shorter than `every`. The
/// ONE monetize is asked again only to notice the day it IS fed (a re-lay
/// that carries the key), which minutes cover.
pub fn unfed_wait(n: u32, every: Duration) -> Duration {
let wait = UNFED_FIRST.saturating_mul(1u32 << n.saturating_sub(1).min(16));
wait.min(UNFED_MAX).max(every)
}
/// The first wait after the ONE monetize refused this console's key.
pub const UNFED_FIRST: Duration = Duration::from_secs(30);
/// The longest.
pub const UNFED_MAX: Duration = Duration::from_secs(300);
/// Did the ONE monetize refuse this console's KEY (not anything else)?
fn refused_the_key(code: &str) -> bool {
code == "unauthenticated"
}
/// The fallback. Parks while the stream is up; on the way back it re-seeds, so
/// the first fallback poll does not re-announce everything the stream already
/// delivered.
async fn poll_forever(monetize: Monetize, hub: EventHub, every: Duration) {
let mut prev: Option<Snapshot> = None;
let mut reachable = true;
let mut was_streaming = false;
// How many polls in a row the ONE monetize refused this console's key.
let mut unfed = 0u32;
loop {
let streaming = hub.is_streaming();
if streaming != was_streaming {
// Entering polling mode: throw the baseline away so the next poll
// seeds silently. Entering streaming mode: same, so a later fallback
// does not diff against a snapshot from before the stream ran.
prev = None;
was_streaming = streaming;
}
if !streaming {
match poll_once(&monetize).await {
Ok(next) => {
if unfed > 0 {
unfed = 0;
tracing::info!("the ONE monetize takes this console's key now; the feed polls it at its own pace again");
}
if !reachable {
reachable = true;
hub.publish(json!({ "kind": "monetize", "reachable": true, "at_unix_ms": now_ms() }));
}
if let Some(p) = &prev {
for ev in diff(p, &next, now_ms()) {
hub.publish(ev);
}
}
prev = Some(next);
}
Err(e) if refused_the_key(&e.code) => {
unfed = unfed.saturating_add(1);
let wait = unfed_wait(unfed, every);
if unfed == 1 || reachable {
reachable = false;
hub.publish(json!({ "kind": "monetize", "reachable": false, "unfed": true, "message": format!("{UNFED_WORDS} ({})", e.message), "at_unix_ms": now_ms() }));
}
tracing::info!(attempt = unfed, wait_s = wait.as_secs(), "the ONE monetize refused this console's key ({}); asked again in {} s", e.message, wait.as_secs());
// A poke (the stream ending) does not shorten it: the
// stream is refused the same key.
tokio::time::sleep(wait).await;
continue;
}
Err(e) if reachable => {
reachable = false;
hub.publish(json!({ "kind": "monetize", "reachable": false, "message": e.message, "at_unix_ms": now_ms() }));
}
Err(_) => {}
}
}
tokio::select! { _ = tokio::time::sleep(every) => {}, _ = hub.wake.notified() => {} }
}
}
/// The stream driver: hold one `Events.Watch`, fan it out, and hand the feed
/// back to the poller — loudly — whenever it cannot.
async fn stream_forever(monetize: Monetize, hub: EventHub) {
// The resume cursor and the epoch it belongs to. A cursor from a previous
// RUN of monetize is meaningless (sequences restart at 1), so a changed
// epoch resets it rather than asking for a replay of somebody else's
// numbers.
let mut from_seq = 0u64;
let mut epoch = String::new();
let mut backoff = Duration::from_millis(250);
// Which mode the feed has been TOLD about, so a monetize that never serves
// `Events` costs one warning rather than one per reconnect.
let mut announced_polling = false;
// How many reconnects in a row the ONE monetize refused this console's key.
let mut unfed = 0u32;
loop {
let opened = monetize.events().watch(pb::WatchRequest { from_seq, ..Default::default() }).await;
match opened {
Ok(response) => {
backoff = Duration::from_millis(250);
unfed = 0;
let mut stream = response.into_inner();
let mut live = false;
let why = loop {
match stream.next().await {
Some(Ok(ev)) => {
if !live {
live = true;
announced_polling = false;
hub.streaming.store(true, Ordering::Relaxed);
tracing::info!(from_seq, "the console feed is live on monetize's Events.Watch; polling is parked");
hub.set_mode("stream", "live on monetize's Events.Watch");
}
if ev.seq > from_seq {
from_seq = ev.seq;
}
if let Some(pb::event::Payload::StreamOpened(p)) = ev.payload.as_ref() {
if !epoch.is_empty() && p.epoch != epoch {
tracing::warn!(
old_epoch = %epoch,
new_epoch = %p.epoch,
"monetize restarted: its event sequence began again, so the cursor is void and the console re-reads"
);
from_seq = 0;
hub.publish(json!({ "kind": "gap", "dropped": 0, "reason": "monetize restarted", "at_unix_ms": now_ms() }));
}
epoch = p.epoch.clone();
continue;
}
if let Some(pb::event::Payload::Gap(g)) = ev.payload.as_ref() {
tracing::warn!(dropped = g.dropped, "this console fell behind monetize's event ring; the lists re-read");
}
if let Some(item) = event_to_feed(&ev) {
hub.publish(item);
}
}
Some(Err(status)) => break format!("the stream failed: {}", status.message()),
None => break "monetize closed the stream".to_owned(),
}
};
hub.streaming.store(false, Ordering::Relaxed);
// Wake the poller at once rather than after its next tick: the
// window between "the stream died" and "the fallback is running"
// is exactly the window in which the console is silently stale.
hub.poke();
if live || !announced_polling {
announced_polling = true;
tracing::warn!(%why, "the console feed FELL BACK TO POLLING monetize; it is no longer live");
hub.set_mode("polling", &why);
}
}
Err(status) if status.code() == tonic::Code::Unauthenticated => {
hub.streaming.store(false, Ordering::Relaxed);
unfed = unfed.saturating_add(1);
if unfed == 1 {
announced_polling = true;
tracing::warn!(why = %status.message(), "the console feed: {UNFED_WORDS}");
hub.set_mode("unfed", &format!("{UNFED_WORDS} ({})", status.message()));
}
tokio::time::sleep(unfed_wait(unfed, MAX_RECONNECT_BACKOFF)).await;
continue;
}
Err(status) => {
hub.streaming.store(false, Ordering::Relaxed);
hub.poke();
if !announced_polling {
announced_polling = true;
let why = format!("Events.Watch refused: {}", status.message());
tracing::warn!(%why, "the console feed FELL BACK TO POLLING monetize; it is no longer live");
hub.set_mode("polling", &why);
}
}
}
tokio::time::sleep(backoff).await;
backoff = (backoff * 2).min(MAX_RECONNECT_BACKOFF);
}
}
/// `GET /api/events`
pub async fn events(State(state): State<AppState>) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
let rx = state.events.subscribe();
// The remembered mode FIRST, before any delta. A tab opened during an
// outage has to be told it is looking at a polled feed; the transition it
// would otherwise have learned from happened before it existed.
let opening = futures_util::stream::once({
let line = state.events.mode_line();
async move { Ok(Event::default().event("feed").data(line.to_string())) }
});
let deltas = tokio_stream::wrappers::BroadcastStream::new(rx).filter_map(|item| async move {
match item {
Ok(text) => Some(Ok(Event::default().event("feed").data(text))),
// Lagged: this BROWSER missed events (monetize's own gap is a
// different hole, and arrives as a `gap` item of its own). Either
// way the answer is the same — tell it to re-read.
Err(_) => Some(Ok(Event::default().event("feed").data(json!({ "kind": "gap", "at_unix_ms": now_ms() }).to_string()))),
}
});
Sse::new(opening.chain(deltas)).keep_alive(KeepAlive::new().interval(Duration::from_secs(15)))
}
#[cfg(test)]
mod tests {
/// A monetize that was not fed this console's key: every call
/// answers `Unauthenticated`, counted.
#[derive(Clone, Default)]
struct Unfed(Arc<std::sync::atomic::AtomicUsize>);
#[tonic::async_trait]
impl pb::products_server::Products for Unfed {
async fn list(&self, _: tonic::Request<pb::ListProductsRequest>) -> Result<tonic::Response<pb::ListProductsResponse>, tonic::Status> {
self.0.fetch_add(1, Ordering::SeqCst);
Err(tonic::Status::unauthenticated("the credential did not verify against any control principal of this monetize"))
}
}
#[tonic::async_trait]
impl pb::events_server::Events for Unfed {
type WatchStream = std::pin::Pin<Box<dyn futures_util::Stream<Item = Result<pb::Event, tonic::Status>> + Send>>;
async fn watch(&self, _: tonic::Request<pb::WatchRequest>) -> Result<tonic::Response<Self::WatchStream>, tonic::Status> {
self.0.fetch_add(1, Ordering::SeqCst);
Err(tonic::Status::unauthenticated("the credential did not verify against any control principal of this monetize"))
}
}
/// ★ **An estate that refuses this console's key is asked less and less,
/// and is named for what it is** (stinkies control-refusal-spam, the REAL
/// test2 2026-10-01: the shared console polled test2 every 5 s with a key
/// test2 was not fed, 7 WARN a minute on test2's front). Polling and the
/// `Events.Watch` reconnects back off from 30 s doubling to 5 min, and the
/// feed says the ONE monetize was "not fed this console's key — re-lay the
/// ONE box".
#[tokio::test]
async fn an_estate_that_refuses_the_consoles_key_is_backed_off_and_named() {
assert_eq!(unfed_wait(1, Duration::from_secs(5)), Duration::from_secs(30));
assert_eq!(unfed_wait(2, Duration::from_secs(5)), Duration::from_secs(60));
assert_eq!(unfed_wait(9, Duration::from_secs(5)), Duration::from_secs(300), "at most five minutes");
let estate = Unfed::default();
let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let at = l.local_addr().unwrap();
tokio::spawn(
tonic::transport::Server::builder()
.add_service(pb::products_server::ProductsServer::new(estate.clone()))
.add_service(pb::events_server::EventsServer::new(estate.clone()))
.serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(l)),
);
let m = Monetize::connect_lazy(&format!("http://{at}"), "a-bearer-this-estate-does-not-take").unwrap();
let hub = EventHub::spawn(m, Duration::from_millis(100));
let mut rx = hub.subscribe();
tokio::time::sleep(Duration::from_secs(4)).await;
let calls = estate.0.load(Ordering::SeqCst);
assert!(calls <= 4, "an unfed estate was asked {calls} times in 4 s (a poll every 100 ms, the stream's reconnects)");
let mut said = Vec::new();
while let Ok(l) = rx.try_recv() {
said.push(l);
}
// The ONE model: the console reads ONE monetize, never an estate's.
assert!(said.iter().any(|l| l.contains("the ONE monetize was not fed this console") && l.contains("re-lay the ONE box")), "the feed must name it: {said:?}");
assert!(!said.iter().any(|l| l.contains("this estate's monetize")), "no estate runs a monetize of its own: {said:?}");
assert!(hub.mode_line()["message"].as_str().unwrap_or("").contains("not fed this console"), "{}", hub.mode_line());
}
use super::*;
#[test]
fn the_diff_names_changed_facts_new_open_attempts_and_closed_ones() {
let mut a = Snapshot::default();
a.facts.insert(("gunnar".into(), "t1".into()), "free|free|None|operator".into());
a.open.insert("gunnar/t2/2026-09-03".into(), 2);
let mut b = a.clone();
assert!(diff(&a, &b, 1).is_empty(), "nothing changed, nothing said");
b.facts.insert(("gunnar".into(), "t1".into()), "10gb-se|paid|Some(5)|payment:invoice:gunnar/t1/2026-09-04".into());
b.open.remove("gunnar/t2/2026-09-03");
b.open.insert("gunnar/t3/2026-09-04".into(), 1);
let evs = diff(&a, &b, 7);
assert_eq!(evs.len(), 3, "{evs:?}");
assert_eq!(evs[0]["kind"], "entitlement");
assert_eq!(evs[0]["state"], "paid");
assert_eq!(evs[0]["source"], "payment:invoice:gunnar/t1/2026-09-04");
assert_eq!((evs[1]["reference"].as_str(), evs[1]["open"].as_bool()), (Some("gunnar/t3/2026-09-04"), Some(true)));
assert_eq!((evs[2]["reference"].as_str(), evs[2]["open"].as_bool()), (Some("gunnar/t2/2026-09-03"), Some(false)));
}
/// (r) Every payload becomes the line the console renders, with its own
/// values in it. Asserting the FIELDS, not that translation returned
/// `Some`: a translator that emitted `{"kind":"ledger"}` for every event
/// would satisfy an `is_some()` test and show an operator a list of blanks.
#[test]
fn every_event_payload_translates_to_the_line_the_console_renders() {
let e = pb::Event::new(
4,
111,
pb::event::Payload::EntitlementChanged(pb::EntitlementChanged {
product: "gunnar".into(),
tenant: "team/sub".into(),
order: "gunnar/team/sub/2026-09-04+pack_bytes=10737418240".into(),
from_state: pb::State::Paid as i32,
had_previous: true,
to_state: pb::State::Grace as i32,
source: "ladder".into(),
pushed: false,
}),
);
let v = event_to_feed(&e).expect("an entitlement change is a feed line");
assert_eq!(v["kind"], "entitlement");
assert_eq!((v["product"].as_str(), v["tenant"].as_str(), v["order"].as_str()), (Some("gunnar"), Some("team/sub"), Some("gunnar/team/sub/2026-09-04+pack_bytes=10737418240")));
assert_eq!((v["state"].as_str(), v["from_state"].as_str()), (Some("grace"), Some("paid")), "both sides of the move");
assert_eq!((v["source"].as_str(), v["pushed"].as_bool(), v["at_unix_ms"].as_u64()), (Some("ladder"), Some(false), Some(111)));
// No previous fact: `from_state` is EMPTY, not "unknown" and not "free".
let mut first = e.clone();
if let Some(pb::event::Payload::EntitlementChanged(c)) = first.payload.as_mut() {
c.had_previous = false;
c.from_state = pb::State::Unspecified as i32;
}
assert_eq!(event_to_feed(&first).unwrap()["from_state"], "");
let e = pb::Event::new(
5,
222,
pb::event::Payload::LedgerAppended(pb::LedgerAppended {
reference: "gunnar/team/sub/2026-09-04".into(),
product: "gunnar".into(),
tenant: "team/sub".into(),
seq: 2,
at_unix_ms: 999,
step: "Captured".into(),
open: true,
}),
);
let v = event_to_feed(&e).unwrap();
assert_eq!(
(v["kind"].as_str(), v["reference"].as_str(), v["step"].as_str(), v["seq"].as_u64(), v["open"].as_bool(), v["at_unix_ms"].as_u64()),
(Some("ledger"), Some("gunnar/team/sub/2026-09-04"), Some("Captured"), Some(2), Some(true), Some(999)),
"the row's own time, not the event's"
);
let e = pb::Event::new(
6,
333,
pb::event::Payload::UsageMeasured(pb::UsageMeasured {
product: "gunnar".into(),
tenant: "team/sub".into(),
meters: vec![pb::Meter { name: "pack_bytes".into(), value: 3 << 30 }],
measured_at_unix_ms: 444,
}),
);
let v = event_to_feed(&e).unwrap();
assert_eq!((v["kind"].as_str(), v["tenant"].as_str()), (Some("usage"), Some("team/sub")));
assert_eq!(v["meters"][0]["name"], "pack_bytes");
assert_eq!(v["meters"][0]["value"].as_u64(), Some(3 << 30));
let e = pb::Event::new(
0,
555,
pb::event::Payload::Gap(pb::EventGap { dropped: 9, last_delivered_seq: 3, resume_seq: 13, epoch: "e".into() }),
);
let v = event_to_feed(&e).expect("a gap must reach the browser; swallowing it is the whole failure mode");
assert_eq!((v["kind"].as_str(), v["dropped"].as_u64(), v["resume_seq"].as_u64()), (Some("gap"), Some(9), Some(13)));
// A REFUSED CALLBACK. The one that has to reach the browser during an
// outage where nothing else on this feed moves.
let e = pb::Event::new(
7,
666,
pb::event::Payload::CallbackRefused(pb::CallbackRefused {
at_unix_ms: 777,
kind: "unverified".into(),
vendor: "stripe".into(),
verified: false,
detail: "stripe: webhook timestamp 1789384242 outside tolerance (300s from now 1789391442)".into(),
body_bytes: 2048,
refusal_seq: 42,
}),
);
let v = event_to_feed(&e).expect("a refused callback is a feed line");
assert_eq!((v["kind"].as_str(), v["refusal"].as_str(), v["vendor"].as_str()), (Some("callback_refused"), Some("unverified"), Some("stripe")));
assert_eq!(v["verified"].as_bool(), Some(false), "whose words `detail` holds is a FIELD, not a guess off the prose");
assert_eq!(v["detail"].as_str(), Some("stripe: webhook timestamp 1789384242 outside tolerance (300s from now 1789391442)"), "monetize's sentence, whole");
assert_eq!((v["body_bytes"].as_u64(), v["refusal_seq"].as_u64(), v["at_unix_ms"].as_u64()), (Some(2048), Some(42), Some(777)), "the refusal's own time, not the event's");
// A PAYMENT FOUND AT THE VENDOR. The money behind the refusal above,
// and the line that says whether monetize could place it.
let e = pb::Event::new(
8,
888,
pb::event::Payload::PaidAtVendor(pb::PaidAtVendorCase {
vendor: "stripe".into(),
session: "cs_live_a1MHA6".into(),
reference: "pi_1".into(),
paid_at_unix_ms: 1_789_470_600_000,
amount_minor: 12_500,
currency: "sek".into(),
tax_minor: 2_500,
livemode: true,
metadata: [("plan".to_string(), "gunnar/10gb-1m".to_string())].into(),
first_seen_unix_ms: 1_789_471_500_000,
state: "needs-operator".into(),
reason: "plan \"gunnar/10gb-1m\" is not sold here".into(),
decided_at_unix_ms: 1_789_471_500_000,
}),
);
let v = event_to_feed(&e).expect("a payment found at the vendor is a feed line");
assert_eq!((v["kind"].as_str(), v["vendor"].as_str(), v["session"].as_str()), (Some("paid_at_vendor"), Some("stripe"), Some("cs_live_a1MHA6")));
assert_eq!((v["amount_minor"].as_u64(), v["currency"].as_str()), (Some(12_500), Some("SEK")), "the vendor's amount, the console's spelling of the code");
assert_eq!((v["state"].as_str(), v["reason"].as_str()), (Some("needs-operator"), Some("plan \"gunnar/10gb-1m\" is not sold here")));
assert_eq!((v["paid_at_unix_ms"].as_u64(), v["at_unix_ms"].as_u64()), (Some(1_789_470_600_000), Some(1_789_471_500_000)), "when the money moved, and when monetize saw it");
assert!(v.get("metadata").is_none() && v.get("email").is_none(), "the feed line is the headline; the list carries the labels, and nothing carries the e-mail");
// The prologue is the driver's business, not the browser's.
let e = pb::Event::new(0, 1, pb::event::Payload::StreamOpened(pb::StreamOpened { epoch: "e".into(), ..Default::default() }));
assert_eq!(event_to_feed(&e), None, "an epoch a browser cannot resume from is not a feed line");
}
}