crates/monetize-server/src/reconcile/tests.rs
//! The classification table, on the host, against a vendor double that
//! answers `paid_since` from a list a test fills. The wire (`ReconcileAdmin`)
//! is driven in `tests/reconcile.rs`.
//!
//! Every test here is one row of the table in the module doc, and the first
//! one is the incident: a payment the vendor took while monetize was refusing
//! every webhook, found by asking, handed to the same engine the webhook
//! feeds, and finished when the sink can be reached.
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use monetize_pay::signup::{SignupError, SignupOutcome};
use super::*;
use crate::poller::tick;
use crate::registry::Registry;
use crate::signup::tests::{a_signup, plans, CountingSink, PLAN};
use crate::signup::SignupState;
use crate::testing::CallbackPay;
/// 2026-09-14T11:10:00Z in ms — when the incident's money moved.
const PAID: i64 = 1_789_470_600_000;
/// The poller's first tick that could have found it: a quarter of an hour on.
const NOW: u64 = (PAID + 15 * 60 * 1000) as u64;
const SIGNUP: &[(&str, &str)] = &[("product", "gunnar"), ("plan", PLAN), ("username", "bob")];
struct Rig {
pay: CallbackPay,
sink: Arc<CountingSink>,
clock: Arc<AtomicU64>,
state: Arc<AppState>,
}
/// A server with the vendor double linked, the counting sink, the plan table
/// a test deployment sells, and a clock the test moves.
fn rig(sink: Arc<CountingSink>) -> Rig {
let pay = CallbackPay::new("mockpay-shared-secret");
let mut state = AppState::in_memory(Registry::new().sells("gunnar", pay.clone()));
state.signup_sink = sink.clone();
state.signup_plans = plans().into();
let clock = Arc::new(AtomicU64::new(NOW));
let tick = Arc::clone(&clock);
state.clock = Box::new(move || tick.load(Ordering::SeqCst));
Rig {
pay,
sink,
clock,
state: Arc::new(state),
}
}
impl Rig {
/// The vendor reports `fact` paid — for a checkout THIS box minted (ST-7).
fn minted_paid(&self, fact: monetize_pay::reconcile::PaidAtVendor) {
self.state
.signups
.record_minted(&fact.vendor, &fact.session, None)
.unwrap();
self.pay.report_paid(fact);
}
}
fn sink_calls(r: &Rig) -> u64 {
r.sink.calls.load(Ordering::SeqCst)
}
/// (r) **The incident, and the way out of it.** The vendor took the money;
/// every webhook was refused; monetize holds nothing. The pass finds the
/// session, records the case, and hands it to the SAME signup engine the
/// webhook feeds — so with the sink saying "not now" (the front's gunnar was
/// refusing over the clock) it is PENDING, and the next tick after the sink
/// answers it is CREATED. RED before this module: the vendor was never asked,
/// and the session stayed "not known to us" until Stripe's retry hours later.
#[test]
fn a_paid_session_unknown_to_monetize_becomes_a_pending_signup_and_is_settled_when_the_sink_answers(
) {
let r = rig(CountingSink::answering(Err(SignupError::Unavailable(
"gunnar: Unauthenticated: the credential is 7200000 ms away from this server's clock"
.into(),
))));
r.minted_paid(CallbackPay::paid_fact("cs_live_a1MHA6", PAID, SIGNUP));
assert!(
r.state
.signups
.get("mockpay", "cs_live_a1MHA6")
.unwrap()
.is_none(),
"monetize holds nothing: that is the incident"
);
let seq_before = r.state.bus.last_seq();
let t = tick(&r.state);
assert_eq!(
(t.reconciled, t.unfulfilled),
(1, 1),
"one fact examined, and it was unknown here"
);
// The signup engine has it, from the vendor's own fact, and it is PENDING
// because the sink said "not now" — the webhook's outcome, from the
// webhook's code.
let rec = r
.state
.signups
.get("mockpay", "cs_live_a1MHA6")
.unwrap()
.expect("the reconciler handed it to the engine");
assert!(rec.state.is_pending(), "{:?}", rec.state);
assert_eq!(
(
rec.signup.username.as_str(),
rec.signup.plan.as_str(),
rec.signup.product.as_str()
),
("bob", PLAN, "gunnar")
);
assert_eq!(
(rec.signup.amount.minor, rec.signup.tax.minor),
(12_500, 2_500),
"the vendor's numbers, tax inside the total"
);
assert_eq!(
rec.signup.paid_at_unix_ms, PAID,
"when the money moved, by the vendor's clock"
);
assert_eq!(
rec.signup.email, "buyer@example.test",
"the sink has to reach the customer"
);
assert_eq!(sink_calls(&r), 1);
// The CASE is durable and says it was handed over — and the bus heard it.
let case = r
.state
.reconcile
.get("mockpay", "cs_live_a1MHA6")
.unwrap()
.expect("a case row");
assert_eq!(
case.state,
CaseState::Settled {
at_unix_ms: NOW as i64
}
);
assert_eq!(
(
case.first_seen_unix_ms,
case.paid_at_unix_ms,
case.amount_minor,
case.currency.as_str()
),
(NOW as i64, PAID, 12_500, "sek")
);
assert_eq!(
r.state.bus.last_seq(),
seq_before + 1,
"one PaidAtVendor event for one new case"
);
let stats = r.state.reconcile.stats().unwrap();
assert_eq!(
(
stats.total,
stats.settled,
stats.needs_operator,
stats.last_pass_unix_ms
),
(1, 1, 0, NOW as i64)
);
// The sink comes back (the clock was fixed). The next tick's
// `settle_pending` finishes it — the reconciler does not have to, and
// must not: one engine, one row.
*r.sink.answer.lock().unwrap() = Ok(SignupOutcome::Created {
namespace: "bob".into(),
username: "bob".into(),
});
r.clock.store(NOW + 5 * 60 * 1000, Ordering::SeqCst);
let t = tick(&r.state);
assert_eq!(
t.signups_settled, 1,
"the pending signup was finished by the poller's own retry"
);
assert_eq!(
(t.reconciled, t.unfulfilled),
(1, 0),
"the vendor answered the same fact again, and it is KNOWN now"
);
let rec = r
.state
.signups
.get("mockpay", "cs_live_a1MHA6")
.unwrap()
.unwrap();
assert!(
matches!(rec.state, SignupState::Created { .. }),
"{:?}",
rec.state
);
assert_eq!(
sink_calls(&r),
2,
"once refused, once created; the reconciler added no call of its own"
);
assert_eq!(
r.state.reconcile.stats().unwrap().total,
1,
"still one case"
);
assert_eq!(
r.state.bus.last_seq(),
seq_before + 1,
"and no second event for it"
);
assert_eq!(
r.state
.signups
.vendor_confirmed("mockpay", "cs_live_a1MHA6"),
Some(PAID),
"the vendor's confirmation is noted beside the row"
);
}
/// (r) **A payment monetize already fulfilled is left exactly as it is.** The
/// ordinary day: the webhook arrived, the account was made, and the vendor's
/// fact list — which answers everything in the window, every pass — must not
/// touch it. Sink call count is the proof, and there is no case row.
#[test]
fn a_paid_session_monetize_already_fulfilled_is_untouched() {
let r = rig(CountingSink::creating());
// The webhook path, as it would have run on a good day.
let by_webhook = crate::signup::settle(&r.state, &a_signup("cs_ok", "bob"));
assert!(matches!(by_webhook, SignupState::Created { .. }));
assert_eq!(sink_calls(&r), 1);
r.minted_paid(CallbackPay::paid_fact("cs_ok", PAID, SIGNUP));
let seq_before = r.state.bus.last_seq();
let t = tick(&r.state);
assert_eq!(
(t.reconciled, t.unfulfilled),
(1, 0),
"examined, known, nothing new"
);
assert_eq!(sink_calls(&r), 1, "the sink was not asked again");
assert!(
r.state.reconcile.get("mockpay", "cs_ok").unwrap().is_none(),
"a fulfilled payment is not a case"
);
assert_eq!(r.state.reconcile.stats().unwrap().total, 0);
assert_eq!(r.state.bus.last_seq(), seq_before, "nothing to announce");
assert_eq!(
r.state.signups.list(0).unwrap().len(),
1,
"one payment, one row, still"
);
assert_eq!(
classify(&r.state, &CallbackPay::paid_fact("cs_ok", PAID, SIGNUP)),
Verdict::Fulfilled
);
assert_eq!(
r.state.signups.vendor_confirmed("mockpay", "cs_ok"),
Some(PAID),
"…but the vendor's confirmation is noted"
);
}
/// (r) **A payment monetize cannot place is left for a person, by name, and
/// nothing is created.** Three shapes of "cannot place": a plan nobody sells,
/// a session with no labels of ours at all, a currency monetize cannot file.
/// Each is a `needs-operator` row whose reason says which; none reaches the
/// sink; none makes a signup row. And a purchase session the ledger holds an
/// attempt for is KNOWN — the transaction owns it — so it is not a case.
#[test]
fn a_paid_session_with_foreign_metadata_needs_an_operator_and_nothing_is_created() {
let r = rig(CountingSink::creating());
r.minted_paid(CallbackPay::paid_fact(
"cs_wrong_plan",
PAID,
&[
("product", "gunnar"),
("plan", "gunnar/1tb-forever"),
("username", "bob"),
],
));
r.minted_paid(CallbackPay::paid_fact("cs_unlabelled", PAID + 1, &[]));
let mut usd = CallbackPay::paid_fact("cs_usd", PAID + 2, SIGNUP);
usd.currency = "usd".into();
r.minted_paid(usd);
// A purchase of ours, mid-flight: the ledger holds the attempt.
r.state
.ledger
.append(
monetize::testing::NOW,
monetize::ledger::Step::Reserved {
reference: "gunnar/acme/2026-09".into(),
handle: monetize_pay::ChargeHandle {
vendor: "mockpay".into(),
id: "res-1".into(),
},
purchase: Box::new(monetize::testing::purchase("acme")),
},
)
.unwrap();
r.minted_paid(CallbackPay::paid_fact(
"cs_purchase",
PAID + 3,
&[("reference", "gunnar/acme/2026-09")],
));
let t = tick(&r.state);
assert_eq!(
(t.reconciled, t.unfulfilled),
(4, 3),
"four facts, three cases; the purchase is the ledger's"
);
assert_eq!(sink_calls(&r), 0, "nothing was guessed at");
assert!(
r.state.signups.list(0).unwrap().is_empty(),
"no signup was made from any of them"
);
assert!(
r.state
.reconcile
.get("mockpay", "cs_purchase")
.unwrap()
.is_none(),
"the ledger knows this money"
);
let cases = r.state.reconcile.list(0).unwrap();
assert_eq!(
cases.iter().map(|c| c.session.as_str()).collect::<Vec<_>>(),
vec!["cs_unlabelled", "cs_usd", "cs_wrong_plan"],
"all seen at NOW: ordered by session then"
);
for c in &cases {
assert!(c.state.needs_operator(), "{c:?}");
assert_eq!(c.first_seen_unix_ms, NOW as i64);
}
let reason = |session: &str| match &r
.state
.reconcile
.get("mockpay", session)
.unwrap()
.unwrap()
.state
{
CaseState::NeedsOperator { reason, .. } => reason.clone(),
other => panic!("{session}: {other:?}"),
};
assert!(
reason("cs_wrong_plan").contains("gunnar/1tb-forever")
&& reason("cs_wrong_plan").contains(PLAN),
"names the plan and what IS sold: {}",
reason("cs_wrong_plan")
);
assert!(
reason("cs_unlabelled").contains("no product and plan"),
"{}",
reason("cs_unlabelled")
);
assert!(
reason("cs_usd").contains("\"usd\""),
"names the currency: {}",
reason("cs_usd")
);
let stats = r.state.reconcile.stats().unwrap();
assert_eq!(
(stats.total, stats.needs_operator, stats.settled),
(3, 3, 0)
);
// The wire row carries the state as a WORD and the reason beside it.
let pb = r
.state
.reconcile
.get("mockpay", "cs_usd")
.unwrap()
.unwrap()
.to_pb();
assert_eq!(
(pb.state.as_str(), pb.currency.as_str(), pb.amount_minor),
("needs-operator", "usd", 12_500)
);
assert!(pb.reason.contains("usd"));
assert_eq!(pb.metadata.get("plan").map(String::as_str), Some(PLAN));
}
/// (r) **A second pass creates nothing.** The vendor answers the same fact on
/// every pass whose window covers it; the session id is the key of both
/// stores, so the second pass — and a webhook for the same session landing
/// after it — is one row, one sink call, one event.
#[test]
fn a_second_pass_creates_nothing_and_neither_does_the_late_webhook() {
let r = rig(CountingSink::creating());
r.minted_paid(CallbackPay::paid_fact("cs_once", PAID, SIGNUP));
let seq_before = r.state.bus.last_seq();
let first = tick(&r.state);
assert_eq!((first.reconciled, first.unfulfilled), (1, 1));
let created_at = r
.state
.reconcile
.get("mockpay", "cs_once")
.unwrap()
.unwrap();
for _ in 0..3 {
r.clock.fetch_add(5 * 60 * 1000, Ordering::SeqCst);
let again = tick(&r.state);
assert_eq!(
(again.reconciled, again.unfulfilled),
(1, 0),
"examined again, and known"
);
}
assert_eq!(sink_calls(&r), 1, "one payment, one account");
assert_eq!(r.state.signups.list(0).unwrap().len(), 1);
assert_eq!(r.state.reconcile.list(0).unwrap().len(), 1);
assert_eq!(
r.state
.reconcile
.get("mockpay", "cs_once")
.unwrap()
.unwrap(),
created_at,
"the row is not re-decided or re-stamped"
);
assert_eq!(
r.state.bus.last_seq(),
seq_before + 1,
"one event, not one per tick"
);
// Stripe's retry finally lands for the same session: the engine's own
// idempotency answers what became of it and calls nothing.
let late = crate::signup::settle(&r.state, &a_signup("cs_once", "bob"));
assert!(matches!(late, SignupState::Created { .. }), "{late:?}");
assert_eq!(sink_calls(&r), 1, "the late webhook made no second account");
// And the store's claim is write-once even when asked directly.
let mut dup = created_at.clone();
dup.state = CaseState::NeedsOperator {
reason: "would overwrite".into(),
at_unix_ms: 0,
};
assert_eq!(
r.state.reconcile.claim(&dup).unwrap(),
Claimed::Already(Box::new(created_at.clone()))
);
assert_eq!(
r.state
.reconcile
.get("mockpay", "cs_once")
.unwrap()
.unwrap(),
created_at
);
}
/// (r) **The window trails the LAST PASS, not the clock**, and the first pass
/// reaches back the first window. A monetize that was down for a week asks
/// for the week; one that ticks every five minutes asks for the margin behind
/// its last pass. And a vendor that answers "cannot be asked" costs nothing:
/// no log line, no stamp, no error.
#[test]
fn the_window_trails_the_last_pass_and_the_first_pass_reaches_back_the_first_window() {
assert_eq!(since_for(None, 1_000), 1_000 - FIRST_WINDOW_MS);
assert_eq!(since_for(Some(500), 1_000), 500 - MARGIN_MS);
const {
assert!(MARGIN_MS >= 24 * HOUR_MS, "Stripe filters by the session's creation and a session lives 24 h: the margin must cover it")
};
let r = rig(CountingSink::creating());
tick(&r.state);
assert_eq!(
r.pay.asked_since(),
vec![NOW as i64 - FIRST_WINDOW_MS],
"never asked before: the first window"
);
assert_eq!(
r.state.reconcile.last_pass("mockpay").unwrap(),
Some(NOW as i64)
);
// A week passes with monetize down, and in the middle of it somebody pays.
// The next pass asks from the last one it finished, minus the margin —
// the whole week, and then some — so the payment is found. A window that
// trailed the CLOCK by 48 h would have missed it for good.
let week = 7 * 24 * HOUR_MS;
r.minted_paid(CallbackPay::paid_fact(
"cs_in_the_gap",
NOW as i64 + week / 2,
SIGNUP,
));
r.clock.store((NOW as i64 + week) as u64, Ordering::SeqCst);
let t = tick(&r.state);
assert_eq!(
r.pay.asked_since()[1],
NOW as i64 - MARGIN_MS,
"from the last pass, not from now"
);
assert_eq!(
(t.reconciled, t.unfulfilled),
(1, 1),
"the payment from the middle of the outage"
);
assert_eq!(
r.state.reconcile.last_pass("mockpay").unwrap(),
Some(NOW as i64 + week)
);
// The pass after that trails the catch-up pass, and asks nothing older.
r.clock.fetch_add(5 * 60 * 1000, Ordering::SeqCst);
let t = tick(&r.state);
assert_eq!(r.pay.asked_since()[2], NOW as i64 + week - MARGIN_MS);
assert_eq!(
(t.reconciled, t.unfulfilled),
(0, 0),
"the gap's payment is now older than the margin, and it is already a row"
);
// The open vendors answer the default. A registry with one of those and
// nothing else reconciles nothing and says nothing.
let quiet =
AppState::in_memory(Registry::new().sells("gunnar", monetize::testing::MockPay::new()));
let t = tick(&quiet);
assert_eq!((t.reconciled, t.unfulfilled), (0, 0));
assert_eq!(
quiet.reconcile.last_pass("mockpay").unwrap(),
None,
"a vendor that cannot be asked is not stamped as asked"
);
assert_eq!(quiet.reconcile.stats().unwrap().last_pass_unix_ms, 0);
}
/// (r) **The list is newest-first by when monetize saw it, with the ceiling,
/// and an empty store is empty** — not a row of zeroes.
#[test]
fn the_list_is_newest_first_and_bounded() {
let r = rig(CountingSink::creating());
assert!(r.state.reconcile.list(0).unwrap().is_empty());
assert_eq!(r.state.reconcile.stats().unwrap(), CaseStats::default());
for i in 0..3u64 {
r.minted_paid(CallbackPay::paid_fact(
&format!("cs_{i}"),
PAID + i as i64,
&[],
));
r.clock.store(NOW + i * 60_000, Ordering::SeqCst);
tick(&r.state);
}
assert_eq!(
r.state
.reconcile
.list(0)
.unwrap()
.iter()
.map(|c| c.session.as_str())
.collect::<Vec<_>>(),
vec!["cs_2", "cs_1", "cs_0"]
);
assert_eq!(
r.state.reconcile.list(2).unwrap().len(),
2,
"limit keeps the newest"
);
assert_eq!(r.state.reconcile.list(2).unwrap()[0].session, "cs_2");
}
/// ★ ST-7: **a session paid at the vendor that this box never minted is another system's** —
/// the reconciler makes no account for it; it is an operator's case, named.
#[test]
fn a_paid_session_this_box_never_minted_is_an_operators_case_and_no_account() {
let r = rig(CountingSink::answering(Ok(SignupOutcome::Created {
namespace: "eve".into(),
username: "eve".into(),
})));
r.pay
.report_paid(CallbackPay::paid_fact("cs_other_system", PAID, SIGNUP));
let _ = tick(&r.state);
assert_eq!(
sink_calls(&r),
0,
"no account was made for another system's money"
);
assert!(r
.state
.signups
.get("mockpay", "cs_other_system")
.unwrap()
.is_none());
match r
.state
.reconcile
.get("mockpay", "cs_other_system")
.unwrap()
.expect("a case row")
.state
{
CaseState::NeedsOperator { reason, .. } => {
assert!(reason.contains("not minted by this monetize"), "{reason}")
}
other => panic!("an operator's case: {other:?}"),
}
}