crates/monetize-server/src/reconcile.rs
//! **Vendor reconciliation: what the vendor says it took, held against what
//! monetize fulfilled.** The other half of the 2026-09-14 incident.
//!
//! MEASURED that day on the front: an Apple Pay checkout (`cs_live_a1MHA6…`,
//! 11:10 UTC, 125.00 SEK) was `complete` and `paid` at Stripe, the charge
//! `succeeded, captured`. Every webhook for it was refused — the box's clock
//! ran two hours fast, so the signed `t=` fell outside the tolerance — and
//! monetize recorded NOTHING. Hours later Stripe's retry landed, the signup
//! was created as `test3`, and the entitlement push failed for an unrelated
//! reason. Between 11:10 and then, the only trace was a journald line. The
//! customer read "This session is not known to us".
//!
//! [`crate::refusals`] made the REFUSALS visible. It cannot make the PAYMENT
//! visible: a refused delivery is one this server did not believe, and nothing
//! in it may be read back as fact. What can be read back as fact is the
//! vendor's own ledger, over the key this deployment already holds — an
//! answer to a question we asked, authenticated by us, not a POST from the
//! internet. That read is [`monetize_pay::Payment::paid_since`]; this
//! module is what the poller does with the answer.
//!
//! ```text
//! poller tick ──▶ for every linked vendor: paid_since(since)
//! │ Unavailable(CANNOT_BE_ASKED) invoice, sandbox: skip, silently
//! │ Unavailable(anything else) an outage: one false row, next tick asks again
//! ▼ Ok(facts)
//! for every fact ── classify ──▶ table below
//! ```
//!
//! # The classification
//!
//! | the vendor's session is… | verdict | what happens |
//! |---|---|---|
//! | in the signup store, CREATED | **fulfilled** | nothing; the vendor is noted as having confirmed it (`SignupStore::vendor_confirmed`) |
//! | in the signup store, PENDING or REFUSED | **known** | nothing new — it is already on the `Paid signups` panel; the vendor's confirmation is noted beside it |
//! | a purchase the ledger holds an attempt for (`metadata[reference]`) | **known** | nothing — `resume_reference` on the same tick asks the vendor's `status` and finishes it |
//! | already a row here | **seen** | nothing; the session id is the key and a row is written once |
//! | **unknown to monetize**, and it carries a `plan` this deployment sells | **settled** | a durable [`PaidUnfulfilled`] row, and the fact is handed to [`crate::signup::settle`] — the SAME engine the webhook feeds — so it becomes PENDING → CREATED (or a named refusal) exactly as a delivery would have |
//! | **unknown to monetize**, and it does not | **needs operator** | a durable row with the reason: a plan nobody sells, a currency monetize cannot file, no labels at all. Never guessed at, never refunded |
//!
//! Handing an unknown signup to `settle` is the one place this module ACTS,
//! and it may because of where the fact came from: the vendor answered our
//! authenticated read about its own object. A callback that had not verified
//! could never reach here. The engine's own checks (the username rule, the
//! plan, the amount net of tax) still run, unchanged, and a row that fails
//! them is a REFUSED signup on the signups panel — the same outcome the
//! webhook would have produced, from the same code.
//!
//! # Idempotent on `(vendor, session)`, twice over
//!
//! The vendor answers a FACT LIST, not a diff: every pass whose window covers
//! a payment answers it again. Two stores make that a no-op. A session the
//! signup store already holds is classified *known* before anything is
//! written, so a payment that arrived by webhook — or that a previous pass
//! handed to the engine — is never handed over twice; the engine's own
//! `(vendor, session)` key catches the race the other way round, a webhook
//! that lands after this pass created the row. And [`ReconcileStore::claim`]
//! writes a case row only where none is, in one redb transaction, so a
//! process that died between "decided" and "recorded" is caught by the first
//! rule on its next pass.
//!
//! # The window
//!
//! Each vendor is asked from **its last successful pass minus
//! [`MARGIN_MS`]** (48 h), and from **now minus [`FIRST_WINDOW_MS`]** (48 h)
//! the first time. The margin is not for the poll interval — five minutes —
//! it is for the vendor's own semantics: Stripe filters sessions by when the
//! page was MINTED, a customer pays up to 24 h later, so a payment at `t` sits
//! on a session from as early as `t − 24 h`. 48 h covers that twice. And
//! because the window trails the LAST pass rather than the clock, a monetize
//! that was down for a week asks for the whole week when it comes back —
//! which is the other shape of the same incident.
//!
//! # Durable, in its own file, not backed up
//!
//! `reconcile.redb`: the case rows and each vendor's last pass. Its own file
//! for the reason `signups.redb` is separate from the ledger — redb is one
//! writer per file, and a reconcile write must not queue behind the poller's
//! usage transaction.
//!
//! The CASE ROWS are not in [`crate::backup`]: the money is accounted for in
//! the signup store (which IS backed up) the moment a row is settled, and a
//! `needs-operator` row is a fact the vendor will answer again on the next
//! pass after a restore. What a lost case row costs is the `first_seen`
//! column, not a payment.
//!
//! **THE PASS STAMPS ARE**, and that sentence used to say they were not. The
//! argument above finishes "…from a window that trails the restored last-pass
//! stamp", and a restore does not restore a stamp that is not in the backup.
//! With none, the first pass asks from `now − `[`FIRST_WINDOW_MS`], so: the
//! disk dies on Monday, the box is rebuilt on Friday, and every session paid
//! Monday to Wednesday whose webhook never landed is older than 48 h at the
//! restore and is never asked about again, by anything. The customer paid, has
//! no account, and no row anywhere says so. So `passes.jsonl` is in the
//! backup, and [`ReconcileStore::restore_passes`] never moves a window
//! forward.
use std::collections::BTreeMap;
use std::path::Path;
use monetize_pay::reconcile::{PaidAtVendor, CANNOT_BE_ASKED};
use monetize_pay::{Currency, Money, PaidSignup, PayError};
use monetize_proto as pb;
use redb::{Database, ReadableTable, TableDefinition};
use crate::state::AppState;
use crate::store::StoreError;
/// `(vendor, session) → PaidUnfulfilled` as JSON.
const CASES: TableDefinition<(&str, &str), &[u8]> = TableDefinition::new("paid_at_vendor");
/// `vendor → the last pass that finished, unix ms`.
const PASSES: TableDefinition<&str, i64> = TableDefinition::new("reconcile_passes");
const HOUR_MS: i64 = 60 * 60 * 1000;
/// **How far behind the LAST pass each vendor is asked from.** See the
/// module doc: it covers the vendor's session lifetime twice, and it trails
/// the last pass, not the clock.
pub const MARGIN_MS: i64 = 48 * HOUR_MS;
/// **How far back the FIRST pass asks**, when no pass has finished for a
/// vendor: 48 hours. The first run on the front is the morning after the
/// incident, and 48 h reaches back past it.
pub const FIRST_WINDOW_MS: i64 = 48 * HOUR_MS;
/// The wire and JSON spellings of [`CaseState`]. One writer.
pub const SETTLED: &str = "settled";
pub const NEEDS_OPERATOR: &str = "needs-operator";
/// What the reconciler did with a payment it had no row for.
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub enum CaseState {
/// Handed to [`crate::signup::settle`]. What became of it is the signup
/// row for the same `(vendor, session)`; this row only says it was handed
/// over, and when.
Settled { at_unix_ms: i64 },
/// monetize would not guess. `reason` is the operator's sentence — what
/// was missing or unsellable — and the row stays until a person acts on
/// it at the vendor.
NeedsOperator { reason: String, at_unix_ms: i64 },
}
impl CaseState {
pub fn name(&self) -> &'static str {
match self {
CaseState::Settled { .. } => SETTLED,
CaseState::NeedsOperator { .. } => NEEDS_OPERATOR,
}
}
pub fn needs_operator(&self) -> bool {
matches!(self, CaseState::NeedsOperator { .. })
}
}
/// **One payment the vendor took and monetize had no row for**, as it is kept
/// and as it is shown. The vendor's fact minus the e-mail (the sink has it,
/// in the signup row, where it is needed; a console is not a mailing list),
/// plus when this server first saw it and what it decided.
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub struct PaidUnfulfilled {
pub vendor: String,
pub session: String,
pub reference: String,
pub paid_at_unix_ms: i64,
pub amount_minor: u64,
pub tax_minor: u64,
pub currency: String,
pub livemode: bool,
pub metadata: BTreeMap<String, String>,
/// By OUR clock, the pass that found it.
pub first_seen_unix_ms: i64,
pub state: CaseState,
}
impl PaidUnfulfilled {
/// The wire row — the SAME message the bus publishes, so a console that
/// saw it live and one that read the list later render identical fields.
pub fn to_pb(&self) -> pb::PaidAtVendorCase {
let (state, reason, decided_at_unix_ms) = match &self.state {
CaseState::Settled { at_unix_ms } => (SETTLED, String::new(), *at_unix_ms),
CaseState::NeedsOperator { reason, at_unix_ms } => (NEEDS_OPERATOR, reason.clone(), *at_unix_ms),
};
pb::PaidAtVendorCase {
vendor: self.vendor.clone(),
session: self.session.clone(),
reference: self.reference.clone(),
paid_at_unix_ms: self.paid_at_unix_ms,
amount_minor: self.amount_minor,
currency: self.currency.clone(),
tax_minor: self.tax_minor,
livemode: self.livemode,
metadata: self.metadata.clone(),
first_seen_unix_ms: self.first_seen_unix_ms,
state: state.to_owned(),
reason,
decided_at_unix_ms,
}
}
}
/// What the store holds, for the console's header and for a test.
#[derive(Clone, Copy, PartialEq, Eq, Debug, Default)]
pub struct CaseStats {
pub total: u64,
pub needs_operator: u64,
pub settled: u64,
/// The latest finished pass over any vendor; 0 when none has.
pub last_pass_unix_ms: i64,
}
/// What [`ReconcileStore::claim`] found.
#[derive(Clone, PartialEq, Eq, Debug)]
pub enum Claimed {
/// No row held this `(vendor, session)`; it does now.
Fresh,
/// A row was already there — written by an earlier pass — and this is it.
Already(Box<PaidUnfulfilled>),
}
/// **The case rows and each vendor's last pass.** See the module doc for why
/// it is its own file and why it is not backed up.
pub struct ReconcileStore {
db: Database,
}
impl ReconcileStore {
/// The most rows [`ReconcileStore::list`] returns when the caller names
/// no limit. A console page, not an export.
pub const LIST_CEILING: usize = 500;
pub fn open(path: &Path) -> Result<Self, StoreError> {
Self::init(Database::create(path)?)
}
pub fn in_memory() -> Result<Self, StoreError> {
Self::init(Database::builder().create_with_backend(redb::backends::InMemoryBackend::new())?)
}
fn init(db: Database) -> Result<Self, StoreError> {
let txn = db.begin_write()?;
txn.open_table(CASES)?;
txn.open_table(PASSES)?;
txn.commit()?;
Ok(Self { db })
}
/// **Write the row if none holds its key; report the one that does.** One
/// redb transaction, for the reason `SignupStore::record_pending` gives:
/// a check-then-write with a gap is two rows for one payment.
pub fn claim(&self, case: &PaidUnfulfilled) -> Result<Claimed, StoreError> {
let txn = self.db.begin_write()?;
let found = {
let mut t = txn.open_table(CASES)?;
let held: Option<Vec<u8>> = t.get((case.vendor.as_str(), case.session.as_str()))?.map(|v| v.value().to_vec());
match held {
Some(bytes) => serde_json::from_slice::<PaidUnfulfilled>(&bytes).ok().map(Box::new).map(Claimed::Already),
None => {
let bytes = StoreError::encode("a case", case)?;
t.insert((case.vendor.as_str(), case.session.as_str()), bytes.as_slice())?;
Some(Claimed::Fresh)
}
}
};
txn.commit()?;
// A row that will not parse is not treated as absent — overwriting it
// would erase the only record that a payment was found — so the claim
// fails loudly instead.
found.ok_or_else(|| StoreError::corrupted(format!("paid-at-vendor row {}:{} will not parse", case.vendor, case.session)))
}
pub fn get(&self, vendor: &str, session: &str) -> Result<Option<PaidUnfulfilled>, StoreError> {
let txn = self.db.begin_read()?;
let t = txn.open_table(CASES)?;
Ok(t.get((vendor, session))?.and_then(|v| serde_json::from_slice(v.value()).ok()))
}
/// **Every case, newest first** by when monetize first saw it, at most
/// `limit` (0 ⇒ [`Self::LIST_CEILING`]). A full scan, for the reason
/// `SignupStore::list` gives: redb orders by the key, and a second
/// time-keyed table would be two copies of every row.
pub fn list(&self, limit: usize) -> Result<Vec<PaidUnfulfilled>, StoreError> {
let limit = if limit == 0 { Self::LIST_CEILING } else { limit.min(Self::LIST_CEILING) };
let mut out = self.all()?;
out.sort_by(|a, b| b.first_seen_unix_ms.cmp(&a.first_seen_unix_ms).then_with(|| a.session.cmp(&b.session)));
out.truncate(limit);
Ok(out)
}
fn all(&self) -> Result<Vec<PaidUnfulfilled>, StoreError> {
let txn = self.db.begin_read()?;
let t = txn.open_table(CASES)?;
let mut out = Vec::new();
for entry in t.iter()? {
let (_, v) = entry?;
if let Ok(row) = serde_json::from_slice::<PaidUnfulfilled>(v.value()) {
out.push(row);
}
}
Ok(out)
}
pub fn stats(&self) -> Result<CaseStats, StoreError> {
let rows = self.all()?;
let needs_operator = rows.iter().filter(|r| r.state.needs_operator()).count() as u64;
let txn = self.db.begin_read()?;
let t = txn.open_table(PASSES)?;
let mut last = 0i64;
for entry in t.iter()? {
let (_, v) = entry?;
last = last.max(v.value());
}
Ok(CaseStats { total: rows.len() as u64, needs_operator, settled: rows.len() as u64 - needs_operator, last_pass_unix_ms: last })
}
/// When `vendor` was last asked to the end, or `None` when never.
pub fn last_pass(&self, vendor: &str) -> Result<Option<i64>, StoreError> {
let txn = self.db.begin_read()?;
let t = txn.open_table(PASSES)?;
Ok(t.get(vendor)?.map(|v| v.value()))
}
/// **Every vendor's last pass**, in key order — `crate::backup`.
pub fn export_passes(&self) -> Result<Vec<(String, i64)>, StoreError> {
let txn = self.db.begin_read()?;
let t = txn.open_table(PASSES)?;
let mut out = Vec::new();
for entry in t.iter()? {
let (k, v) = entry?;
out.push((k.value().to_owned(), v.value()));
}
Ok(out)
}
/// **Restore the stamps, never moving a window FORWARD.**
///
/// A stamp is where the next pass asks the vendor FROM, so a stamp that is
/// too early costs a few idempotent re-reads of sessions already accounted
/// for and a stamp that is too late costs the payment itself. The earlier
/// of what is held and what the backup carries therefore wins — which also
/// makes this safe to run against a store that has already passed once.
pub fn restore_passes(&self, passes: &[(String, i64)]) -> Result<usize, StoreError> {
let txn = self.db.begin_write()?;
{
let mut t = txn.open_table(PASSES)?;
for (vendor, at) in passes {
let held: Option<i64> = t.get(vendor.as_str())?.map(|v| v.value());
let keep = held.map_or(*at, |h| h.min(*at));
t.insert(vendor.as_str(), keep)?;
}
}
txn.commit()?;
Ok(passes.len())
}
pub fn set_last_pass(&self, vendor: &str, at_unix_ms: i64) -> Result<(), StoreError> {
let txn = self.db.begin_write()?;
txn.open_table(PASSES)?.insert(vendor, at_unix_ms)?;
txn.commit()?;
Ok(())
}
}
/// **Where a pass asks `vendor` from.** The last pass minus the margin, or the
/// first window behind `now`. Pure, so the window is a thing a test pins.
pub fn since_for(last_pass: Option<i64>, now: i64) -> i64 {
match last_pass {
Some(t) => t - MARGIN_MS,
None => now - FIRST_WINDOW_MS,
}
}
/// What one vendor fact turned out to be. The table in the module doc, as a
/// value the tests and the log line read.
#[derive(Clone, PartialEq, Eq, Debug)]
pub enum Verdict {
/// A signup row exists and is CREATED.
Fulfilled,
/// A signup row exists and is not (pending, refused), or the ledger holds
/// an attempt for the reference. Already visible; nothing new.
Known,
/// A case row already held this session.
Seen,
/// Unknown to monetize until now; a case row was written with this state.
Recorded(CaseState),
}
/// **The vendor's fact as the signup the webhook would have handed up**, or
/// the reason it cannot be one. Pure: the plan table in, a sentence out.
///
/// The three things it requires are the three the engine cannot proceed
/// without: `product` and `plan` (or it is not a signup at all — a purchase
/// names a `reference`, and one with neither is a session this deployment
/// never labelled), a plan `[signups.<product>.plans]` sells (the engine would refuse it
/// by name, but a refusal is final and this row is better left for a person
/// who can look at the vendor's page), and a currency monetize can file. The
/// username is NOT required: an empty one goes to the engine and comes back
/// as the same renameable refusal a webhook produces, which is the customer's
/// way out and not the operator's.
pub fn signup_of(fact: &PaidAtVendor, plans: &BTreeMap<String, crate::config::SignupPlan>) -> Result<PaidSignup, String> {
let label = |k: &str| fact.metadata.get(k).map(|s| s.trim().to_owned()).unwrap_or_default();
let (product, plan) = (label("product"), label("plan"));
if product.is_empty() || plan.is_empty() {
let keys: Vec<&str> = fact.metadata.keys().map(String::as_str).collect();
return Err(format!("carries no product and plan, so it is not a signup, and no reference the ledger holds; its labels are {keys:?}"));
}
if !plans.contains_key(&plan) {
let known: Vec<&str> = plans.keys().map(String::as_str).collect();
return Err(format!("plan {plan:?} is not sold here; [signups.<product>.plans] names {known:?}"));
}
let Some(currency) = Currency::from_code(&fact.currency) else {
return Err(format!("paid in {:?}, which is neither SEK nor EUR; this deployment cannot file it", fact.currency));
};
Ok(PaidSignup {
vendor: fact.vendor.clone(),
session: fact.session.clone(),
reference: fact.reference.clone(),
email: fact.email.clone(),
username: label("username"),
product,
plan,
amount: Money { minor: fact.amount_minor, currency },
tax: Money { minor: fact.tax_minor, currency },
paid_at_unix_ms: fact.paid_at_unix_ms,
customer_country: fact.customer_country.clone(),
})
}
/// **One vendor fact against monetize's stores.** The table in the module doc.
pub fn classify(state: &AppState, fact: &PaidAtVendor) -> Verdict {
let now = state.now() as i64;
// 1. The signup store: a row, of any state, means this payment reached
// the engine — by webhook or by an earlier pass — and the engine's own
// idempotency owns it from here.
match state.signups.get(&fact.vendor, &fact.session) {
Ok(Some(rec)) => {
state.signups.note_vendor_confirmed(&fact.vendor, &fact.session, fact.paid_at_unix_ms);
return if matches!(rec.state, crate::signup::SignupState::Created { .. }) { Verdict::Fulfilled } else { Verdict::Known };
}
Ok(None) => {}
Err(e) => {
// Cannot tell; must not write a case on a store that will not
// answer, or a payment the engine holds becomes a second row.
eprintln!("monetize-server: reconcile {}: signup store: {e}", fact.key());
return Verdict::Known;
}
}
// 2. The ledger: a purchase session carries `metadata[reference]`, and an
// attempt for it — open or closed — means the transaction knows this
// money. An OPEN one is finished by `resume_reference` on the same
// tick, which asks the vendor's `status` for the handle it holds.
if let Some(reference) = fact.metadata.get("reference").map(|s| s.trim()).filter(|s| !s.is_empty()) {
// A private Gunnar's money (`estate/<co>`, its renewals and plan
// changes) is the estate job's, captured AT THE ORDER (wiring trace
// A16/A23): known while that order is — never an operator case.
if let Some(company) = reference.strip_prefix("estate/").and_then(|r| r.split('/').next()).filter(|c| !c.is_empty()) {
if state.estates.get(company).is_some() {
return Verdict::Known;
}
}
match state.ledger.attempt(reference) {
Ok(Some(_)) => return Verdict::Known,
Ok(None) => {}
Err(e) => {
eprintln!("monetize-server: reconcile {}: ledger: {e}", fact.key());
return Verdict::Known;
}
}
}
// 3. A case row from an earlier pass. Checked before deciding so that a
// needs-operator row is not re-decided (and re-announced) every tick.
match state.reconcile.get(&fact.vendor, &fact.session) {
Ok(Some(_)) => return Verdict::Seen,
Ok(None) => {}
Err(e) => {
eprintln!("monetize-server: reconcile {}: store: {e}", fact.key());
return Verdict::Seen;
}
}
// 4. UNKNOWN AT MONETIZE. This is the case.
let decided = match signup_of(fact, &crate::signup::plans_with(state, fact.metadata.get("plan").map(String::as_str).unwrap_or(""))).and_then(|signup| {
// ★ ST-7: only a session this box minted is settled; another system's money on the
// same account is an operator's case, never an account made here.
match state.signups.minted(&signup.vendor, &signup.session) {
Ok(Some(_)) => Ok(signup),
Ok(None) => Err(format!("{} was not minted by this monetize (another system's checkout on the same account?)", fact.key())),
Err(e) => Err(format!("the signup store could not say who minted {}: {e}", fact.key())),
}
}) {
Ok(signup) => {
// The SAME engine the webhook feeds: the checks, the flight, the
// sink, the store, the log lines. It is idempotent on the session,
// so a webhook that lands after this finds the row and does nothing.
let outcome = crate::signup::settle(state, &signup);
skidbladnir::functional_status(
"monetize-server/reconcile",
&fact.vendor,
true,
&format!("{} paid at the vendor and unknown here; handed to the signup engine -> {outcome:?}", fact.key()),
);
CaseState::Settled { at_unix_ms: now }
}
Err(reason) => {
skidbladnir::functional_status("monetize-server/reconcile", &fact.vendor, false, &format!("{} paid at the vendor and unknown here; NEEDS AN OPERATOR: {reason}", fact.key()));
eprintln!(
"monetize-server: reconcile {} — {} paid at the vendor, unknown here, and monetize will not guess: {reason} [the console reads ReconcileAdmin.ListPaidAtVendor]",
fact.key(),
monetize_pay::major_words(fact.amount_minor, &fact.currency)
);
CaseState::NeedsOperator { reason, at_unix_ms: now }
}
};
let case = PaidUnfulfilled {
vendor: fact.vendor.clone(),
session: fact.session.clone(),
reference: fact.reference.clone(),
paid_at_unix_ms: fact.paid_at_unix_ms,
amount_minor: fact.amount_minor,
tax_minor: fact.tax_minor,
currency: fact.currency.clone(),
livemode: fact.livemode,
metadata: fact.metadata.clone(),
first_seen_unix_ms: now,
state: decided.clone(),
};
match state.reconcile.claim(&case) {
Ok(Claimed::Fresh) => {
state.bus.publish(pb::event::Payload::PaidAtVendor(case.to_pb()));
Verdict::Recorded(decided)
}
// Between step 3 and here — only another process on the same file
// could do that, and redb forbids it. Answer what is there.
Ok(Claimed::Already(_)) => Verdict::Seen,
Err(e) => {
// Decided (and, for a signup, settled in the signup store — which is
// the store that matters) but not recorded here. Loud, and the next
// pass classifies it Known off the signup row.
skidbladnir::functional_status("monetize-server/reconcile", &fact.vendor, false, &format!("{} decided {} but not recorded: {e}", fact.key(), decided.name()));
eprintln!("monetize-server: reconcile {} was decided {decided:?} and the store refused it: {e}", fact.key());
Verdict::Recorded(decided)
}
}
}
/// What one pass over every vendor did. The two counts on the poll line.
#[derive(Clone, Copy, PartialEq, Eq, Debug, Default)]
pub struct Reconciled {
/// Vendor facts examined — every payment every vendor answered.
pub examined: usize,
/// Facts that were UNKNOWN at monetize and became case rows this pass,
/// settled or needs-operator alike.
pub unfulfilled: usize,
}
/// **One pass: every linked vendor asked, every answer classified.** Called
/// from the poller tick, on its blocking thread, after `settle_pending` and
/// `retry_pushes` — so a signup this pass hands to the engine and a sink that
/// says "not now" meet on the NEXT tick's `settle_pending`, not this one's.
pub fn pass(state: &AppState) -> Reconciled {
let mut out = Reconciled::default();
let now = state.now() as i64;
for vendor in state.registry.vendors() {
let name = vendor.name;
let since = match state.reconcile.last_pass(name) {
Ok(last) => since_for(last, now),
Err(e) => {
eprintln!("monetize-server: reconcile {name}: last pass: {e}");
continue;
}
};
let facts = match vendor.paid_since(since) {
Ok(v) => v,
// The default answer: this vendor has nothing to be asked. Not a
// log line — `invoice` would print it every five minutes forever.
Err(PayError::Unavailable(why)) if why == CANNOT_BE_ASKED => continue,
Err(e) => {
skidbladnir::functional_status("monetize-server/reconcile", name, false, &format!("paid_since({since}): {e}"));
eprintln!("monetize-server: reconcile {name}: could not ask the vendor: {e}");
continue;
}
};
for fact in &facts {
out.examined += 1;
if let Verdict::Recorded(_) = classify(state, fact) {
out.unfulfilled += 1;
}
}
// The pass finished, to the end of the list: the next one may trail
// it. Written only on success, so a vendor that was down keeps its
// old stamp and is asked from there when it is back.
if let Err(e) = state.reconcile.set_last_pass(name, now) {
eprintln!("monetize-server: reconcile {name}: last pass: {e}");
}
}
out
}
#[cfg(test)]
pub(crate) mod tests;