crates/monetize-server/src/services/products.rs
use std::sync::Arc;
use monetize_pay::Currency;
use monetize_proto::products_server::Products;
use monetize_proto::{self as pb};
use tonic::{Request, Response, Status};
use crate::services::blocking;
use crate::state::AppState;
pub struct ProductsService(pub Arc<AppState>);
/// One product's last known answer, on the wire. `state` is a WORD and not a
/// bool because there are three answers: a product nobody has asked yet is not a
/// product that is down.
fn reachability_to_pb(h: &crate::health::ProductHealth) -> pb::Reachability {
pb::Reachability {
state: h.reachability.name().into(),
checked_at_unix_ms: h.checked_at_unix_ms,
last_up_unix_ms: h.last_up_unix_ms,
detail: h.reachability.detail().into(),
}
}
/// The paid door's last answer, on the wire. A WORD and not a bool for
/// `reachability_to_pb`'s reason, one state further: `unproven` — nobody could
/// ask — is neither a metered build nor a community one, and a console that
/// folded it into either would be making the claim this probe exists to test.
fn metered_door_to_pb(h: &crate::health::ProductHealth) -> pb::MeteredDoor {
let door = h.metered_door_or_unasked();
pb::MeteredDoor {
state: door.name().into(),
checked_at_unix_ms: h.metered_door_at_unix_ms,
detail: door.detail().into(),
}
}
/// **The spend against one cloud's ceiling** — `UI.md` screen 1's last line,
/// the number that stops everything.
///
/// `spent_minor` is the LEDGER's fold for THIS provider, never the cap's JSON:
/// the same `total_provisioned_minor_by` that `main` reconciles the cap from at boot, so
/// what the console shows is rebuildable from the append-only rows and a lost
/// or restored cache cannot show a budget the ledger has already spent. The
/// month window is the one figure only the cap knows, asked through the seam.
///
/// One line per capped cloud, in the ceiling's currency. A cloud NOTHING wraps
/// names no currency, so the fold is taken per currency the ledger could hold
/// and a line is made for each with spend in it — or one empty-currency zero
/// line, so the console still sees a cloud with no ceiling and says so. The
/// registry refuses to start an enabled cloud without a ceiling; this branch
/// exists for the test registries and for honesty, not for production.
fn spend_pb(
cloud: &crate::hot::HotCloud,
ledger: &monetize::Ledger,
) -> Result<Vec<pb::CloudSpend>, Status> {
let provider = cloud.name().to_owned();
// THIS cloud's rows only: the id each `Provisioned` row carries names the
// provider that bought it. See `Ledger::total_provisioned_minor_by`.
let fold = |c: Currency| {
ledger
.total_provisioned_minor_by(&provider, c)
.map_err(|e| Status::internal(format!("ledger fold: {e}")))
};
let month = cloud.spent_this_month_minor();
let line = |currency: &str, spent_minor: u64, ceiling: Option<monetize_cloud::cap::Ceiling>| {
pb::CloudSpend {
provider: provider.clone(),
currency: currency.to_owned(),
spent_minor,
spent_this_month_minor: month,
ceiling_total_minor: ceiling.and_then(|c| c.total_minor),
ceiling_monthly_minor: ceiling.and_then(|c| c.monthly_minor),
}
};
if let Some(ceiling) = cloud.spend_ceiling() {
return Ok(vec![line(
ceiling.currency.code(),
fold(ceiling.currency)?,
Some(ceiling),
)]);
}
const EVERY: [Currency; 2] = [Currency::Eur, Currency::Sek];
let mut out = Vec::new();
for c in EVERY {
let spent = fold(c)?;
if spent > 0 {
out.push(line(c.code(), spent, None));
}
}
if out.is_empty() {
out.push(line("", 0, None));
}
Ok(out)
}
#[tonic::async_trait]
impl Products for ProductsService {
async fn list(
&self,
req: Request<pb::ListProductsRequest>,
) -> Result<Response<pb::ListProductsResponse>, Status> {
// The calling SITE (its estate's own bearer): its own key decides renewal (ST-3).
let site = req
.extensions()
.get::<crate::auth::CallerEstate>()
.map(|e| e.0.clone());
// On the blocking pool now that the ledger is folded here: redb is a
// blocking store, like every other handler that reads it.
blocking(&self.0, move |s| list_on(s, site.as_deref()))
.await
.map(Response::new)
}
}
/// The deployment's price lists on the wire, one per (product, market), in
/// the market's currency — and the one policy, and the hosts.
fn terms_pb(
t: &crate::state::TermsStore,
) -> (
Vec<pb::PriceList>,
pb::Policy,
std::collections::BTreeMap<String, String>,
) {
let mut prices = Vec::new();
for (product, by_market) in &t.prices {
for (market, list) in by_market {
prices.push(pb::PriceList {
product: product.clone(),
market: crate::convert::market_to_pb(*market) as i32,
currency: crate::convert::money_to_pb(monetize_pay::Money {
minor: 0,
currency: list.currency,
})
.currency,
units: list.units.clone(),
});
}
}
let policy = pb::Policy {
period: t.policy.period.name().into(),
grace_days: u32::from(t.policy.grace_days),
retention_days: u32::from(t.policy.retention_days),
};
(prices, policy, t.hosts.clone())
}
/// **What is sold here**, for a call from `site`: a product that site sells offers self-service
/// renewal when the site's own main's key is held here (`[estates.<site>] appliance_key`).
fn list_on(state: &AppState, site: Option<&str>) -> Result<pb::ListProductsResponse, Status> {
let reg = &state.registry;
let site_checks = |product: &str| {
site.is_some_and(|e| {
reg.product_of_estate(e).as_deref() == Some(product)
&& reg.appliance_key_of(e).is_some()
})
};
let products = reg
.products()
.map(|p| pb::ProductInfo {
id: p.id().into(),
meters: p
.meters()
.iter()
.map(|m| pb::MeterInfo {
name: m.name.into(),
meaning: m.meaning.into(),
enforcement: crate::convert::enforcement_to_pb(m.enforcement) as i32,
scope: crate::convert::scope_to_pb(m.scope) as i32,
// The price-list key a cap here is priced by; empty for a
// count that costs nothing to promise.
unit: m.backing.map(|b| b.unit()).unwrap_or_default().into(),
})
.collect(),
// The dot. Last known, from the poller — see `crate::health`.
reachable: Some(reachability_to_pb(&state.health.get(p.id()))),
// **Whether a console may offer self-service renewal at all.** The
// plugin's own answer (`Product::tenant_renewal`), which on gunnar
// is "an appliance verifying key is configured" — asked here rather
// than derived from the config, because the plugin is the half that
// knows whether it could actually check a ticket. Default false, so
// a product nobody configured draws no route.
tenant_renewal: p.tenant_renewal() || site_checks(p.id()),
// **Whether the paid door is there.** Last known, from the poller,
// for the reason the module's cache exists — and never folded into
// the dot above: a community appliance is perfectly reachable.
metered_door: Some(metered_door_to_pb(&state.health.get(p.id()))),
})
.collect();
let mut spend = Vec::new();
for cloud in reg.clouds() {
spend.extend(spend_pb(cloud, &state.ledger)?);
}
let (prices, policy, hosts) = terms_pb(&state.terms.read().unwrap_or_else(|e| e.into_inner()));
Ok(pb::ListProductsResponse {
products,
vendors: reg.vendor_names().into_iter().map(String::from).collect(),
clouds: reg.cloud_names().into_iter().map(String::from).collect(),
spend,
prices,
policy: Some(policy),
hosts,
default_cloud: reg.default_cloud_name().unwrap_or_default().to_owned(),
vendor_modes: vendor_modes(
reg.vendor_names(),
&state
.config
.read()
.unwrap_or_else(|e| e.into_inner())
.vendors,
),
estate_vendors: reg.estate_vendors(),
})
}
/// **Which keys each vendor holds, live or test** — the block's own
/// `livemode`, which the install writes off the key it feeds. Absent from the
/// block is absent here: the console says "not said", never "test".
pub fn vendor_modes(
names: Vec<&str>,
blocks: &std::collections::BTreeMap<String, crate::config::VendorConfig>,
) -> Vec<pb::VendorMode> {
names
.into_iter()
.map(|v| pb::VendorMode {
vendor: v.to_owned(),
livemode: blocks
.get(v)
.and_then(|b| b.settings.get("livemode"))
.and_then(|x| x.as_bool()),
})
.collect()
}
#[cfg(test)]
mod vendor_mode_tests {
use super::*;
/// ★ owner 2026-10-02: the only difference between two estates is WHICH
/// KEYS they hold — read off the vendor block's own `livemode`, and a
/// block that does not say is not said, never "test".
#[test]
fn each_vendor_says_live_or_test_from_its_own_block_and_silence_is_absent() {
let cfg = crate::config::Config::from_toml("[vendors.stripe.gunnar]\nenabled = true\nlivemode = false\n[vendors.invoice.gunnar]\nenabled = true\n").unwrap();
let modes = vendor_modes(
vec!["stripe/gunnar", "invoice/gunnar", "klarna/gunnar"],
&cfg.vendors,
);
let got: Vec<(&str, Option<bool>)> = modes
.iter()
.map(|m| (m.vendor.as_str(), m.livemode))
.collect();
assert_eq!(
got,
[
("stripe/gunnar", Some(false)),
("invoice/gunnar", None),
("klarna/gunnar", None)
]
);
let live = crate::config::Config::from_toml(
"[vendors.stripe.holger]\nenabled = true\nlivemode = true\n",
)
.unwrap();
assert_eq!(
vendor_modes(vec!["stripe/holger"], &live.vendors)[0].livemode,
Some(true)
);
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::testing::{provisioned_rows, NamedCloud};
use monetize::testing::MockCloud;
use monetize_cloud::cap::{Ceiling, SpendCap};
/// **Each cloud's spend line is what THAT cloud bought.** RED before: both
/// lines read the all-provider EUR total, 3 201.
#[test]
fn each_clouds_spend_line_is_its_own_providers_rows() {
let dir =
std::env::temp_dir().join(format!("monetize-products-spend-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let sandbox = SpendCap::new(
NamedCloud("sandbox", MockCloud::new()),
Ceiling::total(1_000, Currency::Eur),
dir.join("s.json"),
)
.unwrap();
let upcloud = SpendCap::new(
NamedCloud("upcloud", MockCloud::new()),
Ceiling::total(2_500, Currency::Eur),
dir.join("u.json"),
)
.unwrap();
// One cloud nothing wraps, which folds per currency instead.
let bare = NamedCloud("scaleway", MockCloud::new());
let state = AppState::in_memory(
crate::registry::Registry::new()
.cloud_capped(sandbox)
.cloud_capped(upcloud)
.cloud(bare),
);
provisioned_rows(&state.ledger, &[("sandbox", 2_070), ("upcloud", 1_131)]);
let listed = list_on(&state, None).unwrap();
let spent: Vec<(String, String, u64)> = listed
.spend
.iter()
.map(|l| (l.provider.clone(), l.currency.clone(), l.spent_minor))
.collect();
assert!(
spent.contains(&("upcloud".into(), "EUR".into(), 1_131)),
"{spent:?}"
);
assert!(
spent.contains(&("sandbox".into(), "EUR".into(), 2_070)),
"{spent:?}"
);
assert!(
spent.contains(&("scaleway".into(), String::new(), 0)),
"a bare cloud that bought nothing shows nothing spent: {spent:?}"
);
}
}
/// R5a: the terms are read by every console page and every quote. A panic that
/// once held them must not make every later request panic. The one writer
/// (a guest's block admitted, guests.rs) inserts a product's prices and host
/// row by row, so the terms a panic leaves hold whole rows.
#[cfg(test)]
mod poisoned_terms_tests {
use super::*;
#[test]
fn poisoned_terms_still_answer_the_product_list() {
let state = AppState::in_memory(crate::registry::Registry::new());
let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _held = state.terms.write();
panic!("a panic while the terms are held");
}));
assert!(state.terms.is_poisoned());
assert!(list_on(&state, None).is_ok());
// The poller's ladder reads the policy off the same terms every tick.
let fact = monetize_product::EntitlementFact {
tenant: monetize_product::TenantId("bob".into()),
plan: String::new(),
state: monetize_product::State::Paid,
paid_until_unix_ms: Some(1),
caps: Default::default(),
source: "operator".into(),
signature: Vec::new(),
issued_unix_ms: None,
issued_signature: Vec::new(),
};
let _ = crate::poller::ladder_move(&state, &fact, 2);
}
}