crates/monetize-server/src/cli.rs
//! **`monetize-server` — the binary's whole behaviour, as a library function**,
//! so that the vetra build in the private `monetize-impl` is the same program
//! with more factories registered: `cli::run(linked)`. The open binary calls
//! `run(Linked::open())`; the vetra one registers the paid impls on top first.
//!
//! ```text
//! monetize-server [--config /etc/monetize/monetize.toml] [--token <bearer>] [--port N]
//! serve; NO bearer = refused (stinky #9)
//! monetize-server --insecure-open-mode-for-tests
//! serve with NO bearer, every call let in: tests only
//! monetize-server --unit render the systemd unit (no root, no bus)
//! monetize-server --install [--user] write + enable + start the unit (skidbladnir)
//! monetize-server --uninstall [--user] stop + disable + erase the data root, keep keys
//! monetize-server --start | --stop | --status
//! monetize-server --key-print [<product>] the Ed25519 public key <product>'s facts verify with (hex; G8: one key per product)
//! monetize-server --check would this server START on the config? (the start's own checks; starts nothing)
//! monetize-server --export DIR write the operational state as text (no secrets)
//! monetize-server --import DIR read it back into EMPTY stores
//! monetize-server --resend-welcome <vendor>:<session> [--force]
//! ask the RUNNING server to send a created signup's welcome again
//! monetize-server --refund <vendor>:<session> --reason TEXT [--amount MINOR] [--force] [--operator-token T]
//! ask the RUNNING server to give a refused signup's money back
//! monetize-server --settle <vendor>:<session> --note TEXT [--operator-token T]
//! close out a refused signup refunded BY HAND at the vendor
//! ```
//!
//! `--export` is the backup half that reuses gunnar's own replication: the output
//! is deterministic text, so committing `DIR` to a git repository pushed to
//! gunnar means gunnar's §11.1 replication carries it to the standby, and an
//! unchanged state produces no commit. It contains NO SECRETS — see
//! `backup`'s module doc for what is left out and why.
//!
//! `--key-print` is what a PRODUCT verifies monetize's facts with. (`--identity-print`, the key
//! monetize signed its calls TO a product with, went in monetize v2.0 T13: the ONE box dials no
//! product.)
//!
//! No clap: the flag set is eight words, and a dependency for eight words is
//! not a dependency worth a build.
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use monetize::{Ledger, Signer};
use skidbladnir::systemd::Scope;
use crate::config::{Config, DEFAULT_CONFIG_PATH};
use crate::registry::{Linked, Registry};
use crate::state::{system_clock, AppState, UsageStore};
use crate::{auth, lifecycle};
#[derive(Default)]
struct Args {
config: Option<PathBuf>,
token: Option<String>,
port: Option<u16>,
verb: Option<String>,
user_scope: bool,
/// The name `--custody-credential` or `--key-print` names.
product: Option<String>,
/// The directory `--export` / `--import` names.
dir: Option<PathBuf>,
/// `--import` only: restore even though the backup was written by a
/// different signing key. A decision, not a convenience — see
/// `backup::BackupError::KeyMismatch`.
allow_key_mismatch: bool,
/// `--resend-welcome`: the `<vendor>:<session>` whose welcome is sent again.
session: Option<String>,
/// `--resend-welcome`: send even if a welcome already went out.
/// `--refund`: refund a CREATED signup too.
force: bool,
/// `--refund` / `--settle`: the operator bearer, over its ladder.
operator_token: Option<String>,
/// [`INSECURE_OPEN_FLAG`]: serve with no bearer. Tests only.
insecure_open: bool,
/// `--refund`: minor units; absent (0) = the whole payment.
amount_minor: u64,
/// `--refund`'s `--reason`, `--settle`'s `--note`.
words: Option<String>,
}
fn parse_args(words: impl IntoIterator<Item = String>) -> anyhow::Result<Args> {
let mut a = Args::default();
let mut it = words.into_iter().peekable();
while let Some(arg) = it.next() {
match arg.as_str() {
"--config" => {
a.config = Some(PathBuf::from(
it.next()
.ok_or_else(|| anyhow::anyhow!("--config needs a path"))?,
))
}
"--token" => a.token = it.next(),
"--port" => {
a.port = Some(
it.next()
.ok_or_else(|| anyhow::anyhow!("--port needs a number"))?
.parse()?,
)
}
"--user" => a.user_scope = true,
v @ ("--export" | "--import") => {
a.dir = Some(PathBuf::from(
it.next()
.ok_or_else(|| anyhow::anyhow!("{v} needs a directory"))?,
));
a.verb = Some(v.to_string());
}
"--allow-key-mismatch" => a.allow_key_mismatch = true,
"--force" => a.force = true,
"--resend-welcome" => {
a.session = Some(it.next().ok_or_else(|| anyhow::anyhow!("--resend-welcome needs <vendor>:<session>, e.g. `--resend-welcome stripe:cs_live_…`"))?);
a.verb = Some(arg);
}
v @ ("--refund" | "--settle") => {
a.session = Some(it.next().ok_or_else(|| {
anyhow::anyhow!("{v} needs <vendor>:<session>, e.g. `{v} stripe:cs_live_…`")
})?);
a.verb = Some(v.to_string());
}
"--amount" => {
a.amount_minor = it
.next()
.ok_or_else(|| {
anyhow::anyhow!(
"--amount needs minor units, e.g. `--amount 12500` for 125,00"
)
})?
.parse()?
}
v @ ("--reason" | "--note") => {
a.words = Some(
it.next()
.ok_or_else(|| anyhow::anyhow!("{v} needs the words, quoted"))?,
)
}
"--operator-token" => a.operator_token = it.next(),
INSECURE_OPEN_FLAG => a.insecure_open = true,
"--custody-credential" => {
a.product = Some(it.next().ok_or_else(|| anyhow::anyhow!("--custody-credential needs a name, e.g. `--custody-credential dns.cloudflare < zone-and-token.json`"))?);
a.verb = Some(arg);
}
"--key-print" => {
if it.peek().is_some_and(|w| !w.starts_with("--")) {
a.product = it.next();
}
a.verb = Some(arg);
}
v @ ("--unit" | "--install" | "--uninstall" | "--start" | "--stop" | "--status"
| "--check" | "--export-now" | "--version" | "--help") => a.verb = Some(v.to_string()),
other => anyhow::bail!("unknown argument {other:?}; try --help"),
}
}
Ok(a)
}
/// The binary. `linked` is what this build can construct — see
/// [`crate::registry::Linked`]; everything the config enables is looked up there.
pub fn run(linked: Linked) -> anyhow::Result<()> {
run_args(linked, std::env::args().skip(1))
}
/// [`run`] on the words `args` (the program name NOT among them) — the server
/// as a role of another binary: the solo monetize box runs it as
/// `monetize-admin server --config …`, a process of its own (G6).
pub fn run_args(linked: Linked, args: impl IntoIterator<Item = String>) -> anyhow::Result<()> {
let args = parse_args(args)?;
let config_path = args
.config
.clone()
.unwrap_or_else(|| PathBuf::from(DEFAULT_CONFIG_PATH));
let mut cfg = if config_path.exists() {
Config::load(&config_path)?
} else {
Config::default()
};
if let Some(p) = args.port {
cfg.grpc_addr.set_port(p);
}
let (token, whence) = auth::resolve(args.token.as_deref(), &cfg.token_file_path());
let scope = if args.user_scope {
Scope::User
} else {
Scope::System
};
match args.verb.as_deref() {
Some("--help") => {
println!("{}", HELP);
return Ok(());
}
Some("--version") => {
println!("{}", monetize_proto::AGENT);
return Ok(());
}
Some("--unit") => {
print!(
"{}",
lifecycle::unit_text(
&std::env::current_exe()?,
&config_path,
&cfg,
scope,
&lifecycle::secret_env_with(&token, &linked, &cfg.product_config_envs())
)
);
return Ok(());
}
Some("--install") => return install(&config_path, &cfg, scope, &token, &linked),
Some("--uninstall") => {
let report =
lifecycle::lifecycle(&std::env::current_exe()?, &config_path, &cfg, scope, &[])
.uninstall()?;
println!("{report:?}");
return Ok(());
}
Some("--start") => {
return lifecycle::lifecycle(&std::env::current_exe()?, &config_path, &cfg, scope, &[])
.start()
}
Some("--stop") => {
return lifecycle::lifecycle(&std::env::current_exe()?, &config_path, &cfg, scope, &[])
.stop()
}
Some("--status") => {
let lc =
lifecycle::lifecycle(&std::env::current_exe()?, &config_path, &cfg, scope, &[]);
println!("active={} healthy={}", lc.status()?, lc.health()?);
return Ok(());
}
Some("--export") => {
let Some(dir) = args.dir.clone() else {
anyhow::bail!("--export needs a directory")
};
let signer = load_or_mint_signer(&cfg.signing_key_path())?;
// redb is single-process. If the service is running it holds this
// lock, and the raw error ("Database already open") does not tell an
// operator what to do about it — so say it here, with the way out.
let ledger = Ledger::open(&cfg.ledger_path()).map_err(|e| {
anyhow::anyhow!(
"{e}\n\nIf monetize-server is RUNNING it holds the ledger's lock, and redb is \
single-process — a second process cannot read it. Export from the running \
service instead, which needs no downtime:\n \
Ledger.Export over gRPC (writes to backup_dir, default {})\n\
This CLI verb is for the offline cases: recovery, or before a first start.",
cfg.backup_dir_path().display()
)
})?;
let usage = UsageStore::open(&cfg.usage_path())?;
let signups = crate::signup::SignupStore::open(&cfg.signup_path())?;
let reconcile = crate::reconcile::ReconcileStore::open(&cfg.reconcile_path())?;
let nonpay = crate::nonpay::NonpayJournal::open(cfg.nonpay_path())?;
let stores = crate::backup::Stores {
ledger: &ledger,
usage: &usage,
signups: &signups,
reconcile: &reconcile,
nonpay: &nonpay,
};
let m = crate::backup::export(&dir, &stores, &hex(&signer.public_key()))?;
println!(
"exported {} ledger row(s), {} usage reading(s), {} signup(s) and {} nonpay row(s) to {} — NO SECRETS, but the signups carry customers' e-mail addresses (signing key stays at {})",
m.ledger_rows,
m.usage_readings,
m.signups,
m.nonpay_rows,
dir.display(),
cfg.signing_key_path().display()
);
return Ok(());
}
Some("--import") => {
let Some(dir) = args.dir.clone() else {
anyhow::bail!("--import needs a directory")
};
let signer = load_or_mint_signer(&cfg.signing_key_path())?;
std::fs::create_dir_all(&cfg.data_dir)?;
let ledger = Ledger::open(&cfg.ledger_path())?;
let usage = UsageStore::open(&cfg.usage_path())?;
let signups = crate::signup::SignupStore::open(&cfg.signup_path())?;
let reconcile = crate::reconcile::ReconcileStore::open(&cfg.reconcile_path())?;
let nonpay = crate::nonpay::NonpayJournal::open(cfg.nonpay_path())?;
let stores = crate::backup::Stores {
ledger: &ledger,
usage: &usage,
signups: &signups,
reconcile: &reconcile,
nonpay: &nonpay,
};
let m = crate::backup::import(
&dir,
&stores,
&hex(&signer.public_key()),
args.allow_key_mismatch,
)?;
println!(
"imported {} ledger row(s), {} usage reading(s), {} signup(s) and {} nonpay row(s) from {}",
m.ledger_rows,
m.usage_readings,
m.signups,
m.nonpay_rows,
dir.display()
);
return Ok(());
}
// **Seal one of this server's own credentials into the custody**
// (`dns.cloudflare`: `{"zone_id": …, "api_token": …}` on stdin). Never
// echoed; the answer names the credential and its size only.
Some("--custody-credential") => {
let name = args.product.clone().unwrap_or_default();
let mut plain = Vec::new();
std::io::Read::read_to_end(&mut std::io::stdin(), &mut plain)?;
if name.starts_with("dns.") {
monetize_cloud::dns::DnsCredential::from_json(&plain)
.map_err(|e| anyhow::anyhow!("{name}: {e}"))?;
}
let custody = crate::custody::Custody::open(cfg.data_dir.join("custody"))
.map_err(|e| anyhow::anyhow!(e))?;
custody
.seal_credential(&name, &plain)
.map_err(|e| anyhow::anyhow!(e))?;
plain.iter_mut().for_each(|b| *b = 0);
println!(
"sealed the credential {name} ({} bytes) into {}",
plain.len(),
custody.dir().display()
);
return Ok(());
}
Some("--resend-welcome") => {
let Some(handle) = args.session.clone() else {
anyhow::bail!("--resend-welcome needs <vendor>:<session>")
};
println!("{}", resend_welcome(&cfg, &token, &handle, args.force)?);
return Ok(());
}
Some(verb @ ("--refund" | "--settle")) => {
let Some(handle) = args.session.clone() else {
anyhow::bail!("{verb} needs <vendor>:<session>")
};
let (operator, _) = auth::resolve_operator(
args.operator_token.as_deref(),
&cfg.operator_token_file_path(),
);
if operator.is_empty() {
anyhow::bail!(
"{verb} takes the OPERATOR bearer and none is configured: write one to {} (0600) or set ${}, and restart the server so it reads it",
cfg.operator_token_file_path().display(),
crate::config::OPERATOR_TOKEN_ENV
);
}
let words = args.words.clone().unwrap_or_default();
println!(
"{}",
refund_or_settle(
&cfg,
&operator,
verb,
&handle,
args.amount_minor,
&words,
args.force
)?
);
return Ok(());
}
// **Would this config start?** The parse and the terms a start reads
// first (the price lists, the default market) — and nothing else: no
// store is opened, no guest loaded, no port bound. The front on vm1
// (gunnar-front, which runs this binary as its own program and links
// no monetize crate) asks this before it writes a block or a merged
// config, so a malformed one is refused by name instead of stopping
// the running server at its next load.
Some("--check") => {
// The start's own checks (check_config), not a parse alone: what
// would STOP the start is an error here, what it would skip is said.
for line in check_config(&cfg, &linked)? {
println!("start would say: {line}");
}
println!(
"ok: {} is a config this server starts on",
config_path.display()
);
return Ok(());
}
Some("--key-print") => {
let root = load_or_mint_signer(&cfg.signing_key_path())?;
println!("{}", key_print(&cfg, root, args.product.as_deref())?);
return Ok(());
}
// **Readable while the service runs**, unlike `--export`: the queue is
// a flat journal, not a redb file, so a second process reading it takes
// no lock away from the server that is appending to it. It opens
// nothing else — an operator asking who is overdue must not have to
// stop the server to find out.
// **The SCHEDULABLE export**, and the reason it has to exist.
//
// `--export` opens `ledger.redb` in this process, and redb is
// single-process: while `monetize-server.service` runs — and it is
// `Restart=always` on the front — the lock is already taken and the
// verb refuses, correctly, telling the caller to use `Ledger.Export`
// instead. So a systemd timer pointed at `--export` would fail every
// tick, for ever, while looking like a closed gap. The server's own
// timer (`backup_every_minutes`, gap G17) is the schedule; this is the
// verb for an export NOW, outside it.
//
// It takes no directory. `ExportBackupRequest` is empty and the
// server's own `backup_dir` decides, which is the property the config
// key was given deliberately: "an operator chooses this once, here",
// rather than letting any holder of the bearer write anywhere the
// service account can.
Some("--export-now") => {
println!("{}", export_now(&cfg, &token)?);
return Ok(());
}
_ => {}
}
if args.insecure_open {
if !token.trim().is_empty() {
anyhow::bail!("ALERT server.bearer-refused monetize-server: {INSECURE_OPEN_FLAG} and a bearer from {} — a configured bearer is never served open; drop the flag", whence.describe());
}
return serve_insecure_open_mode_for_tests(cfg, &config_path, &linked);
}
eprintln!("monetize-server: bearer from {}", whence.describe());
serve(cfg, &config_path, token, &linked)
}
/// **The one way to serve with NO bearer** — every gRPC call let in
/// unauthenticated. Tests only; never in a unit, never on a box.
pub const INSECURE_OPEN_FLAG: &str = "--insecure-open-mode-for-tests";
/// **No bearer, no server** (stinky #9): an empty bearer is what
/// `nornir_service::auth_interceptor` reads as OPEN, so a start whose bearer
/// is missing, empty or whitespace is refused — one ALERT that names where it
/// was looked for (never a bearer: there is none).
fn refuse_without_bearer(token: &str, cfg: &Config) -> anyhow::Result<()> {
if token.trim().is_empty() {
anyhow::bail!(
"ALERT server.bearer-missing monetize-server: NO bearer — refusing to serve: --token, ${} and the token file {} are all missing or empty, and an empty bearer lets every gRPC call in unauthenticated; write a bearer to {} (0600) and start again ({INSECURE_OPEN_FLAG} is for tests)",
crate::config::TOKEN_ENV,
cfg.token_file_path().display(),
cfg.token_file_path().display()
);
}
Ok(())
}
fn install(
config_path: &Path,
cfg: &Config,
scope: Scope,
token: &str,
linked: &Linked,
) -> anyhow::Result<()> {
let exe = std::env::current_exe()?;
if let Some(tree) = skidbladnir::systemd::enclosing_cargo_target_dir(&exe) {
anyhow::bail!("refusing to install a unit around {} — it is inside the build tree {}; install the binary first", exe.display(), tree.display());
}
std::fs::create_dir_all(&cfg.data_dir)?;
let signer = load_or_mint_signer(&cfg.signing_key_path())?;
eprintln!(
"monetize-server: signing key {} (public {})",
cfg.signing_key_path().display(),
hex(&signer.public_key())
);
// ORDER: create the data root and mint into it FIRST, then the account, then
// hand the whole root over. The account's home is not the data root (see
// `config::SERVICE_HOME`), so `ensure_system_user` re-owns the home only —
// whatever this install just minted needs `own_data_dir` to reach the account
// that will read it, and it must run AFTER the minting or a root-owned 0600
// `signing.key` is left under a monetize-owned directory.
#[cfg(unix)]
if matches!(scope, Scope::System) {
let (uid, gid) = lifecycle::ensure_service_user()?;
eprintln!(
"monetize-server: user {} = {uid}:{gid}, home {}",
crate::config::SERVICE_USER,
crate::config::SERVICE_HOME
);
lifecycle::own_data_dir(cfg, uid, gid)?;
eprintln!(
"monetize-server: data root {} owned by {uid}:{gid}",
cfg.data_dir.display()
);
}
lifecycle::install(
&exe,
config_path,
cfg,
scope,
&lifecycle::secret_env_with(
token,
linked,
&[cfg.product_config_envs(), linked.vendor_instance_envs(cfg)].concat(),
),
)?;
println!(
"installed {} ({scope:?}) listening on {}",
lifecycle::SERVER_BIN,
cfg.grpc_addr
);
Ok(())
}
fn load_or_mint_signer(path: &Path) -> anyhow::Result<Signer> {
if path.exists() {
return Ok(Signer::from_path(path)?);
}
let s = Signer::generate();
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
s.write_to(path)?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?;
}
eprintln!(
"monetize-server: minted a new signing key at {}",
path.display()
);
Ok(s)
}
/// `--key-print [<product>]`: the key `product`'s facts verify under (G8).
fn key_print(cfg: &Config, root: Signer, product: Option<&str>) -> anyhow::Result<String> {
let served: Vec<&String> = cfg.products.keys().collect();
let id = match (product, served.as_slice()) {
(Some(p), _) => p.to_string(),
(None, [one]) => (*one).clone(),
(None, _) => anyhow::bail!("--key-print <product>: each product signs with its own key (G8), so say whose (served: {served:?})"),
};
if !cfg.products.contains_key(&id) {
anyhow::bail!("--key-print {id}: no [products.{id}] in this config (served: {served:?})");
}
Ok(hex(&cfg.keyring(root)?.product(&id).public_key()))
}
fn hex(bytes: &[u8]) -> String {
bytes.iter().map(|b| format!("{b:02x}")).collect()
}
/// **The server itself, as a library call** — what `run` does once the verb
/// is "serve": open the stores, build the registry, start the poller, the
/// guest watcher and the estate jobs, and serve gRPC + the webhook on
/// `cfg.grpc_addr` until ctrl-c. Blocks on its own tokio runtime, so a host
/// that embeds monetize (the gunnar front medium, which is ONE PID-1 program
/// with no process to spawn) calls it on a thread of its own.
///
/// A missing, empty or whitespace `token` is REFUSED before anything is
/// opened (stinky #9): see [`refuse_without_bearer`]. Open mode is
/// [`serve_insecure_open_mode_for_tests`], by name.
pub fn serve(
cfg: Config,
config_path: &Path,
token: String,
linked: &Linked,
) -> anyhow::Result<()> {
refuse_without_bearer(&token, &cfg)?;
serve_bearer_checked(cfg, config_path, token, linked)
}
/// [`serve`] with NO bearer: every gRPC call is let in unauthenticated. What
/// [`INSECURE_OPEN_FLAG`] runs, for tests only — said loud, by name.
pub fn serve_insecure_open_mode_for_tests(
cfg: Config,
config_path: &Path,
linked: &Linked,
) -> anyhow::Result<()> {
eprintln!("monetize-server: {INSECURE_OPEN_FLAG} — NO bearer, every gRPC service is OPEN to whoever reaches {}; tests only", cfg.grpc_addr);
serve_bearer_checked(cfg, config_path, String::new(), linked)
}
/// **The ONE box's state, opened as [`serve`] opens it** — the stores under `data_dir`, the
/// registry over `linked`, the per-product keys (G8), every signup sink, the estate guardian
/// with its pulled estates declared — without serving, ticking or arming anything. The
/// level-1 money-flow suite (monetize-impl `money-flow`) drives the production wiring
/// through it; [`serve`] wraps it and starts the loops.
pub fn open_state(cfg: &Config, linked: &Linked) -> anyhow::Result<AppState> {
std::fs::create_dir_all(&cfg.data_dir)?;
let registry = Registry::from_config_with(cfg, linked)?;
// A product guest that brings its estate lifecycle is registered BEFORE the shop is built
// from it (2026-10-17).
for line in crate::estate_order::guest_lifecycle::follow(®istry) {
eprintln!("monetize-server: {line}");
}
// THE TERMS: the price lists (a typo is refused by name here, not read as a
// gift), the one policy, the hosts. No list at all is the open-source
// deployment and starts fine — every order is then free, and `vendors []`
// is a valid deployment. A PRICED list with nobody to take the money is
// refused at ORDER time, by name (`monetize::Refusal::NoVendor`): the
// list is per market and a vendor may be linked for one market and not
// another, so the honest check is the one made when the market is known.
let mut terms = crate::state::TermsStore {
prices: cfg.price_lists()?,
policy: cfg.policy,
nonpay: cfg.nonpay.clone(),
hosts: cfg
.products
.keys()
.map(|p| (p.clone(), cfg.host_of(p)))
.collect(),
default_market: cfg.default_market()?,
};
// WHAT THE START REFUSED is said by name and skipped — never the end of
// the server. Its price list waits with it: the watcher installs the list
// when the product is taken in live.
for r in registry.vendor_start_refusals() {
let line = r.to_string();
eprintln!("monetize-server: {line}");
crate::guests::journal_push(&line);
}
for r in registry.start_refusals() {
let line = r.to_string();
eprintln!("monetize-server: {line}");
crate::guests::journal_push(&line);
if let crate::registry::GuestChange::Refused { id, .. } = r {
terms.prices.remove(id);
}
}
for line in set_aside_unlinked_prices(&mut terms.prices, &|id: &str| {
registry.find_product(id).is_some()
}) {
eprintln!("monetize-server: {line}");
crate::guests::journal_push(&line);
}
for (product, by_market) in &terms.prices {
for (market, list) in by_market {
if !list.is_free() && !registry.vendors().any(|v| v.markets.contains(market)) {
eprintln!(
"monetize-server: WARNING [prices.{product}.{}] charges for capacity and no vendor linked serves that market (vendors: {:?}) — every priced order there will be refused by name until one is",
market.country(),
registry.vendor_names()
);
}
}
}
// Every `[signups.<product>]` WHOLE, plan table included, `product` set
// from the block's name. The server checks a plan's price; only the sink
// knows what a plan GRANTS (`cap_bytes`, `period_days`), and a sink handed
// the block minus its plans refused at start by name
// (monetize-signup-impl, 2026-09-08). Untyped, exactly as a vendor's
// `[vendors.<kind>.<instance>]` is.
let signup_blocks = cfg.signup_blocks()?;
// Every product's plans and their meters, held LIVE (crate::signup::Plans).
let signup_plans = crate::signup::Plans::of(cfg.signup_plans(), cfg.signup_cap_meters());
// A `[signups.<product>] guest = …` is the sink that CREATES that
// product's account (owner 2026-09-26): the host computes and signs the
// entitlement from the plans (read live) with its own key, the guest makes
// the account and delivers it.
let root = load_or_mint_signer(&cfg.signing_key_path())?;
// G8: every product signs with its own key, derived from the root or
// pinned by its block; a pinned key that is not there stops the start.
let keys = Arc::new(cfg.keyring(root.clone())?);
let build_guest: crate::signup::BuildSink = {
let (plans, policy, data_dir, keys) = (
signup_plans.clone(),
cfg.policy,
cfg.data_dir.clone(),
Arc::clone(&keys),
);
std::sync::Arc::new(move |settings: &serde_json::Value| {
let block = format!(
"[signups.{}]",
settings
.get("product")
.and_then(|p| p.as_str())
.unwrap_or("?")
);
let p = plans.clone();
let keys = Arc::clone(&keys);
// The signup's fact is signed with ITS product's key (G8).
let fact_for: monetize_product_wasm::signup::FactFor = Box::new(move |s| {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
crate::signup::signup_fact(
&p.snapshot(),
&policy,
keys.product(&s.product),
s,
now,
&p.cap_meter_of(&s.product),
)
});
let g =
monetize_product_wasm::signup::GuestSignup::build(&data_dir, settings, fact_for)?;
let line = format!(
"signup guest LOADED {block} product {} blake3 {}{}",
g.product(),
g.digest_hex(),
g.signer()
.map(|s| format!(", signed by {s}"))
.unwrap_or_default()
);
eprintln!("monetize-server: {line}");
crate::guests::journal_push(&line);
Ok(std::sync::Arc::new(g) as crate::signup::Sink)
})
};
// Each product's sink: the pull sink when its estate calls monetize
// (`[products.<id>] instance`), else its guest, else the linked sink handed
// its block, else `NoSink` — and `Signup.Start` sells that product none
// until a guest block arrives live. A signup guest that does not load is
// REFUSED by name and that product sells no signup — never the end of the server.
let mut sinks = std::collections::BTreeMap::new();
for (product, settings) in &signup_blocks {
let sink: crate::signup::Sink = if let Some(s) = crate::signup_pull::pulled_sink(
cfg,
product,
®istry.pull_slot(),
&signup_plans,
Arc::clone(&keys),
) {
s
} else if monetize_product_wasm::signup::is_guest_block(settings) {
match build_guest(settings) {
Ok(s) => s,
Err(e) => {
let line = format!("signup guest REFUSED [signup-guest-refused-at-start] [signups.{product}] {e} — {product} sells no signup until a good one arrives");
eprintln!("monetize-server: {line}");
crate::guests::journal_push(&line);
std::sync::Arc::new(crate::signup::NoSink)
}
}
} else {
linked
.build_signup_sink(&cfg.data_dir, settings)
.map_err(|e| anyhow::anyhow!("[signups.{product}] {e}"))?
};
sinks.insert(product.clone(), sink);
}
let signup_hot = crate::signup::HotSink::per_product(sinks);
// The host's relay for customer notices the sink cannot send (g-nonpay,
// g-renewal): the fed `secret.smtp_*` in the secrets dir, re-read per notice.
signup_hot.mail_from(&cfg.secrets_dir());
match signup_hot.host_mailer() {
Ok(Some(m)) => eprintln!("monetize-server: mail: the host's relay is {} — nonpay and renewal notices go through it when the sink sends none", m.relay().describe()),
Ok(None) => eprintln!("monetize-server: mail: no relay (secret.smtp_url / MONETIZE_SMTP_URL) — a customer notice the sink cannot send is recorded NOT SENT"),
Err(e) => eprintln!("monetize-server: mail: REFUSED [mail-relay] {e} — customer notices are recorded NOT SENT until it is fixed"),
}
let signup_reload = std::sync::Arc::new(
crate::signup::SignupReload::new(signup_hot.clone(), build_guest)
.started_from(&signup_blocks),
);
// The whole `[tax]` block, untyped, to whichever jurisdiction is linked —
// the same shape `[signups.<product>]` and `[vendors.<kind>.<instance>]` travel in.
let tax_settings = serde_json::to_value(&cfg.tax).map_err(|e| anyhow::anyhow!("[tax]: {e}"))?;
let home_country = home_country_of(&tax_settings, cfg.default_market()?);
// WHERE EACH SUPPLIER IS, from the operator's own declaration. A vendor or
// cloud that declares nothing is simply absent from the map, and
// `monetize::Books` reads absence as domestic — never as a reverse charge
// nobody wrote down.
let supplier_countries = crate::guests::supplier_countries(cfg);
let pull_slot = registry.pull_slot();
let state = AppState {
ledger: Ledger::open(&cfg.ledger_path())?,
keys,
terms: std::sync::RwLock::new(terms),
backup_dir: Some(cfg.backup_dir_path()),
usage: UsageStore::open(&cfg.usage_path())?,
usage_cursors: Default::default(),
health: crate::health::HealthStore::default(),
// The mock's time travel, when (and only when) the config names one.
clock: match cfg.bench_travel_url.as_deref() {
Some(url) => {
crate::bench::travel_clock(url, std::time::Duration::from_secs(2))
.map_err(|e| anyhow::anyhow!("{e}"))?
.0
}
None => system_clock(),
},
checkout_urls: std::sync::RwLock::new(crate::guests::checkout_urls(cfg)),
changes: std::sync::Mutex::new(Vec::new()),
registry,
bus: std::sync::Arc::new(crate::bus::EventBus::default()),
// **Not opened from a path**: the refused-callback ring is in memory
// and clears on restart, deliberately — see `crate::refusals`. It is
// written by unauthenticated input, so it is the one store here that
// must not be able to touch the disk the ledger is on.
refusals: crate::refusals::RefusalLog::default(),
signups: crate::signup::SignupStore::open(&cfg.signup_path())?,
// Durable: a delivery the vendor was told 200 for is kept before it
// is acted on (`crate::inbox`).
callbacks: crate::inbox::CallbackInbox::open(&cfg.callback_inbox_path())?,
resuming: crate::inbox::Flights::default(),
// Durable, unlike the refusal ring: a row here is the vendor's answer
// to OUR authenticated read, and it must survive a restart because the
// incident it records is one where monetize was not listening.
reconcile: crate::reconcile::ReconcileStore::open(&cfg.reconcile_path())?,
// Durable, its own flat file: a restart that forgot a `Cancelled` row
// would delete a customer who paid.
nonpay: crate::nonpay::NonpayJournal::open(cfg.nonpay_path())?,
// One sink per `[signups.<product>]`, built above: its guest, else the
// linked sink (the same registration seam the vendors use, so this
// file names no private crate), else `NoSink`. They stand in a
// HotSink: the guest watcher swaps a product's sink when its block or
// module changes (crate::signup::SignupReload), because the front's
// monetize starts before the feed that names it.
signup_sink: signup_hot.clone(),
signup_plans: signup_plans.clone(),
signup_reload: Some(signup_reload.clone()),
// `[products.<id>.bundles]`, as declared. Priced per request against
// the market the caller names — see `AppState::bundles`.
bundles: std::sync::RwLock::new(crate::guests::bundles_of(cfg)),
// **The sold-capacity allowance.** Refused at start when it is below
// 100, rather than read as a reserve nobody asked for; absent is the
// default, which is finite by construction — see `crate::oversell`.
signup_max_sold_percents: cfg.signup_max_sold_percents()?,
signup_reservations: Default::default(),
// The jurisdiction, through the same registration seam the vendors and
// the sink use, so this file names no private crate. With none linked and
// none declared it is `monetize::PerCountry`: each buyer country's totals,
// from the country, rate and tax the vendor returned per payment.
tax: linked
.build_tax(&cfg.data_dir, &tax_settings)
.map_err(|e| anyhow::anyhow!("{e}"))?,
// The DEPLOYMENT's own country is the default market's: that is the
// market this server prices in when a request names none, and inventing
// a second "where are we" key that could disagree with it would be a
// second source of truth for one fact.
books: std::sync::RwLock::new(monetize::Books {
home_country,
supplier_countries,
}),
grow: crate::grow::GrowSets::from_config(cfg)?,
estates: crate::estate_order::EstateOrders::from_config_with(cfg, linked)?,
named_users: crate::named_users::NamedUsers::open(
cfg.data_dir.join(crate::named_users::FILE),
)
.map_err(|e| anyhow::anyhow!("named users: {e}"))?,
// The whole file, kept for the estate PICTURE and for nothing else —
// see `AppState::config`. Cloned rather than re-read per request: the
// board must show the estate this process is running under.
config: std::sync::RwLock::new(cfg.clone()),
plugin_jobs: Default::default(),
callback_slots: Default::default(),
control: {
let (keys, said) = crate::control::ControlKeys::from_config(&cfg.control_principals);
for l in said {
eprintln!("monetize-server: {l}");
}
Arc::new(keys)
},
plans: crate::plans::PlanBook::new(cfg.pricing.clone()),
fleet: crate::fleet::FleetState::sharing(pull_slot),
tests: crate::test_purchase::TestRuns::open(
cfg.data_dir.join(crate::test_purchase::RUNS_FILE),
),
};
// THE ESTATE GUARDIAN (5.4): the pull channel's store beside the ledger; it signs the
// orders it queues with the same key the boxes know as `monetize_pubkey`.
{
let store = monetize_guardian::Store::open(&cfg.data_dir.join("fleet.redb"))
.map_err(|e| anyhow::anyhow!("fleet.redb: {e}"))?;
// The provider looks are a slow job behind the tick (`fleet::tick_on`), never inside it.
let gcfg = monetize_guardian::GuardianConfig {
receipts_dir: Some(cfg.data_dir.join("fleet-receipts")),
looks_in_tick: false,
..Default::default()
};
state.fleet.install(monetize_guardian::Guardian::new(
Arc::new(store),
Arc::clone(&state.keys),
gcfg,
));
for g in &linked.guardians {
state.fleet.link(g.product, g.guardian);
}
crate::fleet::declare_pulled(&state, cfg)?;
state
.fleet
.joined_at(cfg.data_dir.join(crate::fleet::JOINED_REL));
}
// The ledger's rows become `Events.Watch` events from here on. `in_memory`
// does the same for the test path; both go through the one method.
state.wire_events();
Ok(state)
}
fn serve_bearer_checked(
cfg: Config,
config_path: &Path,
token: String,
linked: &Linked,
) -> anyhow::Result<()> {
std::fs::create_dir_all(&cfg.data_dir)?;
// **The OPERATOR bearer** — `SignupAdmin.Refund` / `Settle` take it and
// nothing else does (g-refund-verb). Read here and not in `run`, so a host
// that embeds `serve` gets the same door. None ⇒ those verbs are CLOSED.
let (operator, whence) = auth::resolve_operator(None, &cfg.operator_token_file_path());
if operator.is_empty() {
eprintln!(
"monetize-server: no operator bearer ({} or ${}) — SignupAdmin.Refund / Settle are refused to everybody",
cfg.operator_token_file_path().display(),
crate::config::OPERATOR_TOKEN_ENV
);
} else if operator == token {
eprintln!("monetize-server: the operator bearer IS the service bearer — that separates nothing, so SignupAdmin.Refund / Settle are refused to everybody");
} else {
eprintln!(
"monetize-server: operator bearer from {}",
whence.describe()
);
}
let state = Arc::new(open_state(&cfg, linked)?);
// The wallet readings run on their own jobs, and the `server` cloud queues
// a grant past its walls where the estate grows its set when told.
crate::grow::arm(&state);
// The plans' price basis: read from the provider now, kept warm (owner,
// 2026-10-05: cached ≥ 10 minutes, warmed at start, refreshed behind).
crate::plans::arm(&state);
// THE CAP'S CACHE IS CORRECTED FROM THE LEDGER, at every start, before a
// single purchase can be attempted. The cap keeps a small JSON total so it
// need not fold the ledger on every `ensure`; the ledger is append-only and
// records a cost on every purchase, so it is the authority. Without this a
// lost or restored-from-older-backup cache would quietly hand the deployment
// its whole budget again.
for line in reconcile_spend_caps(&state) {
eprintln!("monetize-server: {line}");
}
{
let t = state.terms.read().unwrap_or_else(|e| e.into_inner());
let priced: Vec<String> = t
.prices
.iter()
.flat_map(|(p, m)| {
m.iter().map(move |(k, l)| {
format!(
"{p}/{}{}",
k.country(),
if l.is_free() { " (free)" } else { "" }
)
})
})
.collect();
eprintln!(
"monetize-server: products {:?} vendors {:?} clouds {:?} policy {}/{}d/{}d price lists {:?}{}",
state.registry.product_ids(),
state.registry.vendor_names(),
state.registry.cloud_names(),
t.policy.period.name(),
t.policy.grace_days,
t.policy.retention_days,
priced,
if priced.is_empty() { " — none: every order is free" } else { "" }
);
}
// The server's jera ledger (RA3-5): its one-shot jobs and long duties are rows in
// `<data>/jobs`; one that cannot open is said, and the jobs run unrecorded.
if let Err(e) = crate::jobs::set_ledger(&cfg.data_dir.join("jobs")) {
eprintln!("monetize-server: ALERT jobs.unrecorded the jobs' ledger: {e}; the jobs run, unrecorded");
}
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()?;
rt.block_on(async move {
let listener = crate::bind(cfg.grpc_addr).await?;
eprintln!("monetize-server: listening on {} (gRPC h2c + POST /callback/<vendor>)", cfg.grpc_addr);
// A32-14, stays a task: serve's poller tick on the server's runtime; its handle is aborted at the stop below.
let poll = tokio::spawn(crate::poller::run(state.clone(), Duration::from_secs(cfg.poll_minutes.max(1) * 60)));
let backups = (cfg.backup_every_minutes > 0).then(|| {
eprintln!("monetize-server: backup into {} every {} min", cfg.backup_dir_path().display(), cfg.backup_every_minutes);
// A32-14, stays a task: the backup tick; its handle is aborted at the stop below.
tokio::spawn(crate::backup::run(state.clone(), Duration::from_secs(cfg.backup_every_minutes * 60)))
});
// GUESTS ON THE FLY: a new or rebuilt `.wasm` guest is taken in without a
// restart, through the same gates as at start (`crate::guests`).
let _guest_watch = match crate::guests::watch(state.clone(), config_path.to_path_buf(), cfg.clone(), crate::guests::WATCH_EVERY) {
Ok(h) => {
eprintln!(
"monetize-server: guests on the fly: watching {} and every file its guest blocks name, every {} s",
config_path.display(),
crate::guests::WATCH_EVERY.as_secs()
);
Some(h)
}
Err(e) => {
eprintln!("monetize-server: ALERT guests.watch-unstarted the guest watcher did not start ({e}): a new or rebuilt guest needs a restart");
None
}
};
// THE LAY-ESTATE JOBS (private gunnars): every open order is driven
// from its file at start and every 2 s after — resumed, never restarted.
let estates = state.estates.cfg.is_some().then(|| {
eprintln!("monetize-server: estate orders: {} open order(s) read back", state.estates.all().iter().filter(|r| crate::estate_order::phase::is_open(&r.phase)).count());
// A32-14, stays a task: the estate tick (it queues each drive as a jera job); aborted at the stop below.
tokio::spawn(crate::estate_order::run(state.clone(), Duration::from_secs(2)))
});
// THE ESTATE GUARDIAN's tick (5.4): estates that joined taken, reports read, cases
// moved, orders queued — every 5 s, on THE monetize whatever it sells. It never calls
// into an estate.
// A jera duty (P4b): stopped with the server's other jobs at the clean stop below.
if state.fleet.guardian().is_some() {
eprintln!("monetize-server: estate guardian: {} estate(s) known, ticking every 5 s", state.fleet.guardian().map(|g| g.store.estates().len()).unwrap_or(0));
if let Err(e) = crate::fleet::start(&state, Duration::from_secs(5)) {
eprintln!("monetize-server: ALERT jobs.unstarted {}: the estate guardian does not tick: {e}", crate::fleet::TICK_JOB);
}
}
let shutdown = async {
let _ = tokio::signal::ctrl_c().await;
eprintln!("monetize-server: shutting down");
};
crate::serve_on_with(listener, state, Arc::new(token), Arc::new(operator), shutdown).await?;
// The server's jera jobs (A32-14: the keepers, the guest watch, the drives) end here, and
// nothing new starts after the clean stop.
let jobs = crate::jobs::stop_all(Duration::from_secs(2));
if jobs > 0 {
eprintln!("monetize-server: {jobs} job(s) stopped");
}
if let Some(e) = estates {
e.abort();
}
poll.abort();
if let Some(b) = backups {
b.abort();
}
Ok::<(), anyhow::Error>(())
})
}
const HELP: &str = "monetize-server [--config PATH] [--token BEARER] [--port N] [--user]
--unit render the systemd unit
--install write + enable + start the unit via skidbladnir (root for system scope)
--uninstall stop + disable + erase the data root, keep the keys
--start | --stop | --status
--export-now ask the RUNNING server to write its backup (Ledger.Export) into the
configured `backup_dir`, and print what it wrote. This is the one a
timer can name: `--export` opens the redb in THIS process and the
serving one already holds the lock.
--key-print [<product>]
the public signing key (hex) <product>'s estates verify its facts with —
one key per product (G8); the product may be left out when one is served
--check run the start's own checks (registry, prices, market, [signups.*], [tax]); print ok or the reason; starts nothing
--export DIR write the operational state as deterministic text; NO secrets.
Commit DIR to a git repo pushed to a forge whose own
replication carries it to the standby.
--import DIR read a backup back into EMPTY stores. Refuses a backup written
by a different signing key unless --allow-key-mismatch.
--custody-credential <name>
seal one of this server's own credentials, read from stdin, into
<data_dir>/custody under the custody key — `dns.cloudflare` is
the private estates' DNS writer: {\"zone_id\": …, \"api_token\": …}.
Never echoed; never in monetize.toml or the environment.
--resend-welcome <vendor>:<session> [--force]
a CLIENT verb: asks the running server (grpc_addr, with the
bearer) to send a CREATED signup's welcome mail again. Sends
nothing if one already went out, unless --force. Audited by the
server and the sink.
--refund <vendor>:<session> --reason TEXT [--amount MINOR] [--force]
a CLIENT verb (SignupAdmin.Refund, OPERATOR bearer): give a
REFUSED signup's money back through its vendor. No --amount is
the whole payment. Safe to repeat: a refund whose answer was
lost is finished under the same idempotency key, a done one is
only reported. --force refunds a CREATED signup too (the account
stays). ListOwedBack drops the row once it is DONE.
--settle <vendor>:<session> --note TEXT
a CLIENT verb (SignupAdmin.Settle, OPERATOR bearer): close out a
refused signup you refunded BY HAND at the vendor. Moves no money.
Vendor credentials: set them in the environment of the `--install` and they are
carried into the unit's 0600 EnvironmentFile (never into the 0644 unit), e.g.
MONETIZE_INVOICE_IMPORT_TOKEN=... monetize-server --install
Bearer: --token > $MONETIZE_SERVER_TOKEN > <data_dir>/token > REFUSED (no bearer, no server).
--insecure-open-mode-for-tests serves with NO bearer, every call let in: tests only.
Operator bearer (--refund, --settle; read by the server at start):
--operator-token > $MONETIZE_SERVER_OPERATOR_TOKEN > <data_dir>/operator-token > NONE
(none: the money verbs are refused to everybody).";
/// **`--resend-welcome`: a client of the RUNNING server**, not a second opener of
/// its stores. redb is single-process and the service holds `signups.redb`; the
/// sink that knows the relay and the audit trail lives in that process too. So
/// this is one `SignupAdmin.ResendWelcome` call to `grpc_addr` with the bearer
/// every other operator surface uses.
fn resend_welcome(cfg: &Config, token: &str, handle: &str, force: bool) -> anyhow::Result<String> {
resend_welcome_within(cfg, token, handle, force, OPERATOR_CALL_WITHIN)
}
/// How long an operator verb waits for the RUNNING server: the connect, the h2 handshake and
/// the call together (P4c: only the TCP connect was bounded; a server that took the connection
/// and never answered held the verb forever).
const OPERATOR_CALL_WITHIN: Duration = Duration::from_secs(120);
/// `call` within `within`, or the verb's words that it was not answered.
async fn bounded<T>(
within: Duration,
verb: &str,
call: impl std::future::Future<Output = anyhow::Result<T>>,
) -> anyhow::Result<T> {
match tokio::time::timeout(within, call).await {
Ok(out) => out,
Err(_) => Err(anyhow::anyhow!("{verb}: the running server did not answer within {} s; it may still finish what it was asked — look before asking again", within.as_secs_f64())),
}
}
/// [`resend_welcome`], bounded by `within`.
fn resend_welcome_within(
cfg: &Config,
token: &str,
handle: &str,
force: bool,
within: Duration,
) -> anyhow::Result<String> {
let (vendor, session) = parse_signup_handle(handle)?;
let mut addr = cfg.grpc_addr;
if addr.ip().is_unspecified() {
// A server bound to 0.0.0.0 is reached on loopback from its own box.
addr.set_ip(std::net::IpAddr::from([127, 0, 0, 1]));
}
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
rt.block_on(bounded(within, "--resend-welcome", async move {
let channel = tonic::transport::Endpoint::from_shared(format!("http://{addr}"))?
.connect_timeout(Duration::from_secs(10))
.connect()
.await?;
let mut client = monetize_proto::signup_admin_client::SignupAdminClient::new(channel);
let mut req = tonic::Request::new(monetize_proto::ResendWelcomeRequest {
vendor,
session,
force,
});
if !token.is_empty() {
req.metadata_mut().insert(
"authorization",
format!("Bearer {token}")
.parse()
.map_err(|e| anyhow::anyhow!("the bearer is not a header value: {e}"))?,
);
}
let out = client
.resend_welcome(req)
.await
.map_err(|s| anyhow::anyhow!("{:?}: {}", s.code(), s.message()))?;
Ok(out.into_inner().outcome)
}))
}
/// **`--refund` / `--settle`: a client of the RUNNING server**, the same shape
/// as [`resend_welcome`] — one `SignupAdmin.Refund` / `Settle` call to
/// `grpc_addr` — with the OPERATOR bearer, which is what those two take.
fn refund_or_settle(
cfg: &Config,
operator: &str,
verb: &str,
handle: &str,
amount_minor: u64,
words: &str,
force: bool,
) -> anyhow::Result<String> {
refund_or_settle_within(
cfg,
operator,
verb,
handle,
(amount_minor, words, force),
OPERATOR_CALL_WITHIN,
)
}
/// [`refund_or_settle`], bounded by `within` (owner 2026-11-03: a timeout only; the server's
/// refund and settle are untouched, and a verb given up on says the server may still finish).
/// `ask` is the verb's amount in minor units, its words and its force.
fn refund_or_settle_within(
cfg: &Config,
operator: &str,
verb: &str,
handle: &str,
ask: (u64, &str, bool),
within: Duration,
) -> anyhow::Result<String> {
let (amount_minor, words, force) = ask;
let (vendor, session) = parse_signup_handle(handle)?;
let mut addr = cfg.grpc_addr;
if addr.ip().is_unspecified() {
addr.set_ip(std::net::IpAddr::from([127, 0, 0, 1]));
}
let bearer: tonic::metadata::MetadataValue<tonic::metadata::Ascii> =
format!("Bearer {operator}")
.parse()
.map_err(|e| anyhow::anyhow!("the operator bearer is not a header value: {e}"))?;
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
let out = rt.block_on(bounded(within, verb, async move {
let channel = tonic::transport::Endpoint::from_shared(format!("http://{addr}"))?
.connect_timeout(Duration::from_secs(10))
.connect()
.await?;
let mut client = monetize_proto::signup_admin_client::SignupAdminClient::new(channel);
let status = |s: tonic::Status| anyhow::anyhow!("{:?}: {}", s.code(), s.message());
let out = if verb == "--refund" {
let mut req = tonic::Request::new(monetize_proto::RefundSignupRequest {
vendor,
session,
amount_minor,
reason: words.to_owned(),
force,
});
req.metadata_mut().insert("authorization", bearer);
client.refund(req).await.map_err(status)?
} else {
let mut req = tonic::Request::new(monetize_proto::SettleSignupRequest {
vendor,
session,
note: words.to_owned(),
});
req.metadata_mut().insert("authorization", bearer);
client.settle(req).await.map_err(status)?
};
Ok::<_, anyhow::Error>(out.into_inner())
}))?;
Ok(format!(
"{:?} {} {}{} — {}",
out.state(),
out.refund_id,
out.amount
.as_ref()
.map(|m| major(m.minor))
.unwrap_or_default(),
out.amount
.as_ref()
.and_then(|m| monetize_proto::Currency::try_from(m.currency).ok())
.map(|c| format!(" {}", c.as_str_name()))
.unwrap_or_default(),
out.reason
))
}
/// **Ask the RUNNING server to write its backup**, and say what it wrote.
///
/// One `Ledger.Export` call to `grpc_addr` with the bearer every other
/// operator surface uses — the same shape as [`resend_welcome`], and for the
/// same reason: the stores belong to the serving process and a second one
/// cannot read them.
///
/// The line it prints names the directory, the row counts and the public
/// signing key the manifest recorded. That last one is not decoration: a
/// restore refuses by default into a server holding a different key, so an
/// operator reading a backup's own output can tell whether it is restorable
/// where they mean to restore it.
fn export_now(cfg: &Config, token: &str) -> anyhow::Result<String> {
export_now_within(cfg, token, OPERATOR_CALL_WITHIN)
}
/// [`export_now`], bounded by `within`.
fn export_now_within(cfg: &Config, token: &str, within: Duration) -> anyhow::Result<String> {
let mut addr = cfg.grpc_addr;
if addr.ip().is_unspecified() {
// A server bound to 0.0.0.0 is reached on loopback from its own box.
addr.set_ip(std::net::IpAddr::from([127, 0, 0, 1]));
}
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
rt.block_on(bounded(within, "--export-now", async move {
let channel = tonic::transport::Endpoint::from_shared(format!("http://{addr}"))?
.connect_timeout(Duration::from_secs(10))
.connect()
.await?;
let mut client = monetize_proto::ledger_client::LedgerClient::new(channel);
let mut req = tonic::Request::new(monetize_proto::ExportBackupRequest {});
if !token.is_empty() {
req.metadata_mut().insert(
"authorization",
format!("Bearer {token}")
.parse()
.map_err(|e| anyhow::anyhow!("the bearer is not a header value: {e}"))?,
);
}
let out = client
.export(req)
.await
.map_err(|s| anyhow::anyhow!("{:?}: {}", s.code(), s.message()))?
.into_inner();
Ok(format!(
"exported to {} — {} ledger row(s), {} usage reading(s), signed by {}",
out.dir, out.ledger_rows, out.usage_readings, out.signing_public_key
))
}))
}
/// `stripe:cs_live_…` → `("stripe", "cs_live_…")`. The first `:` splits, because
/// a vendor name never holds one and a session id might.
fn parse_signup_handle(handle: &str) -> anyhow::Result<(String, String)> {
match handle.split_once(':') {
Some((v, s)) if !v.is_empty() && !s.is_empty() => Ok((v.to_owned(), s.to_owned())),
_ => anyhow::bail!("{handle:?} is not <vendor>:<session> (e.g. stripe:cs_live_…)"),
}
}
/// Minor units to `1000.00`. Integer arithmetic, never a float — the console's
/// `fmt::money_amount` does the same on the other side of the wire, and money
/// that has been through an `f64` is money nobody can reconcile.
fn major(minor: u64) -> String {
format!("{}.{:02}", minor / 100, minor % 100)
}
/// **The spend cap's boot line — the credit limit, in units a human has.**
///
/// This is the line an operator reads to answer *what is the worst this
/// deployment can spend?*, and it is load-bearing in a way the rest of the boot
/// output is not: there is no "uncapped" spelling in the config file, so a
/// deployment that means "do not stop my customers" writes a deliberately large
/// `total_minor` and that NUMBER is then the only thing between a runaway loop
/// and an invoice (see `config`'s *"Uncapped is not a thing this file can
/// say"*).
///
/// It used to print `ceiling monthly=None total=Some(100000)`. That is the
/// stored form, and it fails the one job the line has: `100000` is EUR 1000 or
/// EUR 100000 depending on a fact the line did not carry, and `Some(_)` is
/// Rust's word, not an operator's. Wrong by a factor of a hundred, in the
/// direction of "that looks fine", is exactly the mistake this line exists to
/// prevent.
///
/// `left` is the ceiling minus what the LEDGER says is committed — saturating,
/// because a ceiling lowered after a purchase really does go negative and
/// `0.00 left` is the honest reading of it. The month figure comes from the cap
/// itself (`spent_this_month_minor`), which is the only thing that knows the
/// window; `None` means no monthly limit is set, which is the common case and
/// says so in words rather than in punctuation.
/// **Each capped cloud's cache, corrected from the ledger rows THAT CLOUD
/// bought** — one line per cloud, for the boot log.
///
/// Per provider since 2026-09-14. Before, every cap was reconciled from the
/// sum over all providers, so a sandbox rehearsal's play-money was pushed into
/// UpCloud's real ceiling (and the other way round). `reconcile_spend` only
/// ever RAISES a cache (see `SpendCap::reconcile_total`), so a cache that was
/// already raised by the old combined figure stays where it is until an
/// operator corrects the `<name>-spend.json` beside the ledger.
pub fn reconcile_spend_caps(state: &crate::state::AppState) -> Vec<String> {
let mut lines = Vec::new();
for cloud in state.registry.clouds() {
let Some(ceiling) = cloud.spend_ceiling() else {
continue;
};
match state
.ledger
.total_provisioned_minor_by(cloud.name(), ceiling.currency)
{
Ok(total) => match cloud.reconcile_spend(total) {
Err(e) => lines.push(format!("{} spend cap: {e}", cloud.name())),
Ok(()) => lines.push(spend_cap_line(
cloud.name(),
total,
cloud.spent_this_month_minor(),
ceiling,
)),
},
Err(e) => lines.push(format!("ledger total for {}: {e}", cloud.name())),
}
}
lines
}
pub(crate) fn spend_cap_line(
provider: &str,
committed_minor: u64,
month_minor: Option<u64>,
ceiling: monetize_cloud::cap::Ceiling,
) -> String {
let cur = ceiling.currency.code();
let total = match ceiling.total_minor {
Some(t) => format!(
"total {} {cur} ({} left)",
major(t),
major(t.saturating_sub(committed_minor))
),
None => "total not set".to_string(),
};
let monthly = match (ceiling.monthly_minor, month_minor) {
(Some(m), Some(spent)) => format!(
"monthly {} {cur} ({} committed this month)",
major(m),
major(spent)
),
(Some(m), None) => format!("monthly {} {cur}", major(m)),
(None, _) => "monthly not set".to_string(),
};
format!(
"{provider} spend cap — {} {cur} committed per the ledger; ceiling {total}; {monthly}",
major(committed_minor)
)
}
/// **Where this deployment is established, for its books** (g-dk-market):
/// the `[tax] jurisdiction` it declares when it names one — the Danish company
/// files in Denmark whatever market an unnamed order defaults to — else the
/// default market's country, as it always was.
pub fn home_country_of(
tax_settings: &serde_json::Value,
default_market: monetize_pay::Market,
) -> String {
match tax_settings
.get("jurisdiction")
.and_then(|j| j.as_str())
.map(str::trim)
{
Some(j) if j.len() == 2 && j.chars().all(|c| c.is_ascii_uppercase()) => j.to_owned(),
_ => default_market.country().to_owned(),
}
}
/// **What the start would refuse, without starting** (release-3: three start-time
/// config fatals were found only 11 minutes into a KVM mock). The same calls
/// [`serve`] makes before it opens a store — the registry (products, vendors,
/// clouds, their start refusals), the price lists (a product not linked yet is
/// set aside, [`set_aside_unlinked_prices`]), the default market, every
/// `[signups.<product>]` and the `[tax]` block and the tax filing they build — and nothing is written but
/// what the tax factory writes under `cfg.data_dir`. `Ok` carries every line the
/// start would SAY (refused by name and skipped); `Err` is what would STOP it.
/// `monetize-server --check` is its door.
pub fn check_config(cfg: &Config, linked: &Linked) -> anyhow::Result<Vec<String>> {
let registry = Registry::from_config_with(cfg, linked)?;
let mut said: Vec<String> = registry
.vendor_start_refusals()
.iter()
.map(|r| r.to_string())
.collect();
let mut prices = cfg.price_lists()?;
for r in registry.start_refusals() {
said.push(r.to_string());
if let crate::registry::GuestChange::Refused { id, .. } = r {
prices.remove(id);
}
}
said.extend(set_aside_unlinked_prices(&mut prices, &|id: &str| {
registry.find_product(id).is_some()
}));
cfg.default_market()?;
cfg.signup_blocks()?;
cfg.signup_max_sold_percents()?;
let tax_settings = serde_json::to_value(&cfg.tax).map_err(|e| anyhow::anyhow!("[tax]: {e}"))?;
linked
.build_tax(&cfg.data_dir, &tax_settings)
.map_err(|e| anyhow::anyhow!("{e}"))?;
Ok(said)
}
/// **Price lists for products not linked YET, set aside at start** — pure.
/// A product whose plugin SELF-REGISTERS (gunnar's, after monetize serves) is
/// not linked when the start reads `[prices.<id>]`; its list is taken in live
/// when the product is (`guests::apply_guests`' admit_new reads the config's
/// price lists). Returns one line per list set aside; the rest stay.
pub fn set_aside_unlinked_prices<V>(
prices: &mut std::collections::BTreeMap<String, V>,
linked: &dyn Fn(&str) -> bool,
) -> Vec<String> {
let gone: Vec<String> = prices.keys().filter(|id| !linked(id)).cloned().collect();
gone.into_iter()
.map(|id| {
prices.remove(&id);
format!("price list REFUSED [prices-product-not-linked-at-start] [prices.{id}]: no [products.{id}] is linked yet — it is taken in live when the product registers")
})
.collect()
}
#[cfg(test)]
mod tests {
/// ★ P4c (lane mon-shared) RED->GREEN: an operator verb against a server that takes the
/// connection and never answers is bounded (it hung forever: only the TCP connect was).
#[test]
fn an_operator_verb_against_a_silent_server_is_bounded() {
let silent = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let cfg = crate::config::Config {
grpc_addr: silent.local_addr().unwrap(),
..Default::default()
};
let started = std::time::Instant::now();
let within = std::time::Duration::from_millis(300);
for (verb, out) in [
(
"--resend-welcome",
super::resend_welcome_within(&cfg, "t", "stripe:cs_1", false, within),
),
("--export-now", super::export_now_within(&cfg, "t", within)),
(
"--refund",
super::refund_or_settle_within(
&cfg,
"op",
"--refund",
"stripe:cs_1",
(100, "test", false),
within,
),
),
(
"--settle",
super::refund_or_settle_within(
&cfg,
"op",
"--settle",
"stripe:cs_1",
(0, "test", false),
within,
),
),
] {
let e = out.expect_err(verb).to_string();
assert!(e.contains("did not answer within"), "{verb}: {e}");
}
assert!(
started.elapsed() < std::time::Duration::from_secs(10),
"bounded: {:?}",
started.elapsed()
);
drop(silent);
}
/// ★ owner 2026-10-11: nonpay purges after the grace; there is no reclaim
/// queue beside it to read (no legacy).
#[test]
fn the_reclaim_queue_is_gone() {
let e = super::parse_args(["--reclaim-queue".to_string()])
.err()
.expect("refused")
.to_string();
assert!(e.contains("unknown argument"), "{e}");
}
/// **G8: `--key-print <product>` prints that product's key**, never the
/// root's; with one product served the name may be left out, with several
/// it may not — a kit that records "monetize's key" must say whose.
#[test]
fn key_print_names_the_product_whose_key_it_prints() {
let root = || monetize::Signer::from_bytes(&[6u8; 32]);
let two =
crate::config::Config::from_toml("[products.gunnar]\n[products.holger]\n").unwrap();
let derived = |p: &str| super::hex(&monetize::Keyring::derive(&root(), p).public_key());
assert_eq!(
super::key_print(&two, root(), Some("holger")).unwrap(),
derived("holger")
);
assert_ne!(
super::key_print(&two, root(), Some("holger")).unwrap(),
super::hex(&root().public_key()),
"not the root"
);
let e = super::key_print(&two, root(), None)
.unwrap_err()
.to_string();
assert!(
e.contains("--key-print <product>") && e.contains("gunnar") && e.contains("holger"),
"{e}"
);
assert!(
super::key_print(&two, root(), Some("nobody")).is_err(),
"a product this config does not serve"
);
let one = crate::config::Config::from_toml("[products.gunnar]\n").unwrap();
assert_eq!(
super::key_print(&one, root(), None).unwrap(),
derived("gunnar")
);
let args = super::parse_args(["--key-print".to_string(), "holger".to_string()]).unwrap();
assert_eq!(
(args.verb.as_deref(), args.product.as_deref()),
(Some("--key-print"), Some("holger"))
);
let bare = super::parse_args([
"--key-print".to_string(),
"--config".to_string(),
"/x".to_string(),
])
.unwrap();
assert_eq!(
(bare.verb.as_deref(), bare.product.as_deref()),
(Some("--key-print"), None)
);
}
/// ★ release-3 (MEASURED on the dwarf mock 2026-09-29, install-orig-6): the
/// paid front's monetize read `[prices.gunnar]` before gunnar self-registered
/// its plugin, and the start DIED "prices a product that is not linked".
/// ★ `check_config` refuses what the start refuses, by the start's own words:
/// the `[tax]` that stopped vm1's monetize on the mock (install-orig-7).
#[test]
fn check_config_refuses_what_the_start_refuses_and_says_what_it_skips() {
let dir =
std::env::temp_dir().join(format!("monetize-check-config-{}", std::process::id()));
let base = format!("data_dir = {:?}\n[vendors.mock.gunnar]\nenabled = true\n[prices.gunnar.SE]\ndisk_gib_month = 2200\n", dir.display().to_string());
fn build_mock(
_: &monetize_pay::factory::VendorBuild<'_>,
) -> Result<Box<dyn monetize_pay::Payment>, String> {
Ok(Box::new(monetize::testing::MockPay::new()))
}
let mut open = crate::registry::Linked::open();
open.vendor(monetize_pay::factory::VendorFactory {
name: "mock",
config_envs: &[],
build: build_mock,
});
let no_rate =
crate::config::Config::from_toml(&format!("{base}[tax]\njurisdiction = \"DK\"\n"))
.unwrap();
let e = super::check_config(&no_rate, &open).expect_err("the start refuses it");
assert!(format!("{e:#}").contains("needs rate_percent"), "{e:#}");
let ok = crate::config::Config::from_toml(&format!(
"{base}[tax]\njurisdiction = \"DK\"\nrate_percent = 25\nprices = \"exclusive\"\n"
))
.unwrap();
let said = super::check_config(&ok, &open).expect("the start takes it");
assert!(
said.iter().any(|l| l.contains("[prices.gunnar]")),
"{said:?}"
);
}
#[test]
fn a_price_list_for_a_product_not_linked_yet_is_set_aside_and_the_rest_stay() {
let mut prices = std::collections::BTreeMap::from([
("gunnar".to_string(), 1u8),
("acme".to_string(), 2u8),
]);
let said = super::set_aside_unlinked_prices(&mut prices, &|id: &str| id == "acme");
assert_eq!(prices.keys().cloned().collect::<Vec<_>>(), ["acme"]);
assert!(
said.len() == 1
&& said[0].contains("[prices.gunnar]")
&& said[0].contains("prices-product-not-linked-at-start"),
"{said:?}"
);
}
use super::*;
use monetize_cloud::cap::Ceiling;
/// ST-15 (owner: "no legacy", fresh ledger OM-5): **the ledger merge is gone** — no
/// `--ledger-merge`, `--from`, `--rename` words, and the help names none of them.
#[test]
fn the_ledger_merge_is_gone() {
let words = |w: &[&str]| w.iter().map(|s| s.to_string()).collect::<Vec<_>>();
for w in [
&["--ledger-merge"][..],
&["--from", "/b"][..],
&["--rename", "a=b"][..],
] {
assert!(
parse_args(words(w)).is_err(),
"{w:?} is no word of this server"
);
}
assert!(
!HELP.contains("--ledger-merge") && !HELP.contains("--rename"),
"the help names no merge"
);
}
/// **c-dead-front**: a START whose config prices a product not registered
/// yet ended the server (`[prices.gunnar] prices a product that is not
/// linked`, MEASURED on the recharge mock) where the hot reload waits for
/// the product. The list waits, said by name; a linked product keeps its.
#[test]
fn prices_of_a_product_not_linked_yet_wait_and_the_server_starts() {
let mut prices: std::collections::BTreeMap<String, u8> =
[("gunnar".to_string(), 1u8), ("holger".to_string(), 2u8)]
.into_iter()
.collect();
// release-3: recharge's prices_waiting and release-3's set_aside_unlinked_prices
// were the same fix; ONE function stays (the start and check_config use it).
let said = set_aside_unlinked_prices(&mut prices, &|id: &str| id == "holger");
assert_eq!(
prices.keys().cloned().collect::<Vec<_>>(),
vec!["holger".to_string()]
);
assert!(
said.len() == 1
&& said[0].contains("[prices.gunnar]")
&& said[0].contains("taken in live when the product registers"),
"{said:?}"
);
}
use monetize_pay::Currency;
/// **The line an operator checks the credit limit against, in the units a
/// human has.** `total=Some(100000)` is the shape the switch to real
/// UpCloud leans on (see `config`'s *"Uncapped is not a thing this file can
/// say"*): the ONLY thing standing between the deployment and a runaway
/// loop is a number somebody typed, so that number has to be legible at
/// boot. Minor units and `Some(_)` are not legible — 100000 is EUR 1000 or
/// EUR 100000 depending on a fact the line does not carry.
#[test]
fn the_home_country_is_the_tax_jurisdiction_the_estate_declares() {
// g-dk-market: the Danish company files in Denmark while its default
// MARKET (the currency an unnamed order is priced in) stays SE.
assert_eq!(
home_country_of(
&serde_json::json!({ "jurisdiction": "DK" }),
monetize_pay::Market::Se
),
"DK"
);
assert_eq!(
home_country_of(
&serde_json::json!({ "period": "quarterly" }),
monetize_pay::Market::Se
),
"SE",
"no jurisdiction: the default market's country, as before"
);
}
#[test]
fn the_cap_line_states_the_ceiling_in_major_units_and_what_is_left() {
let ceiling = Ceiling::total(100_000, Currency::Eur);
assert_eq!(
spend_cap_line("upcloud", 2_260, None, ceiling),
"upcloud spend cap — 22.60 EUR committed per the ledger; ceiling total 1000.00 EUR (977.40 left); monthly not set"
);
}
/// Twin: the same committed figure against a TIGHT ceiling reads
/// differently, so the sentence is computed and not a template with the
/// numbers pasted in.
#[test]
fn the_cap_line_twin_a_tight_ceiling_says_how_little_is_left() {
let ceiling = Ceiling {
monthly_minor: Some(5_000),
total_minor: Some(2_500),
currency: Currency::Eur,
};
assert_eq!(
spend_cap_line("upcloud", 2_260, Some(1_000), ceiling),
"upcloud spend cap — 22.60 EUR committed per the ledger; ceiling total 25.00 EUR (2.40 left); monthly 50.00 EUR (10.00 committed this month)"
);
}
/// **At boot each cap is corrected from the rows ITS provider bought.**
/// 2 070 cents of sandbox play-money and 1 131 of UpCloud: UpCloud's cap
/// learns 1 131, the sandbox's 2 070. RED before: both learned 3 201.
#[test]
fn boot_reconciles_each_cap_from_its_own_providers_rows() {
use crate::testing::{provisioned_rows, NamedCloud};
use monetize::testing::MockCloud;
use monetize_cloud::cap::SpendCap;
let dir =
std::env::temp_dir().join(format!("monetize-cli-reconcile-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let sandbox = SpendCap::new(
NamedCloud("sandbox", MockCloud::new()),
Ceiling::total(1_000_000, Currency::Eur),
dir.join("sandbox-spend.json"),
)
.unwrap();
let upcloud = SpendCap::new(
NamedCloud("upcloud", MockCloud::new()),
Ceiling::total(2_500, Currency::Eur),
dir.join("upcloud-spend.json"),
)
.unwrap();
let state = crate::state::AppState::in_memory(
crate::registry::Registry::new()
.cloud_capped(sandbox)
.cloud_capped(upcloud),
);
provisioned_rows(
&state.ledger,
&[("sandbox", 230), ("sandbox", 1_840), ("upcloud", 1_131)],
);
let lines = reconcile_spend_caps(&state);
assert_eq!(lines.len(), 2, "{lines:?}");
assert!(
lines
.iter()
.any(|l| l.starts_with("upcloud spend cap — 11.31 EUR committed")),
"{lines:?}"
);
assert!(
lines
.iter()
.any(|l| l.starts_with("sandbox spend cap — 20.70 EUR committed")),
"{lines:?}"
);
// The cache on disk is what the next purchase is admitted against.
let stored = |name: &str| -> u64 {
let v: serde_json::Value =
serde_json::from_slice(&std::fs::read(dir.join(name)).unwrap()).unwrap();
v["total_minor"].as_u64().unwrap()
};
assert_eq!(
stored("upcloud-spend.json"),
1_131,
"sandbox play-money reached upcloud's real ceiling"
);
assert_eq!(stored("sandbox-spend.json"), 2_070);
}
/// A ceiling already exceeded says `0.00 left` rather than underflowing —
/// and this really happens: `reconcile_total` corrects the cache UPWARD
/// from the ledger, so a ceiling lowered after a purchase is exactly this.
#[test]
fn the_cap_line_never_underflows_past_a_ceiling_already_exceeded() {
let ceiling = Ceiling::total(1_000, Currency::Eur);
assert_eq!(
spend_cap_line("upcloud", 4_000, None, ceiling),
"upcloud spend cap — 40.00 EUR committed per the ledger; ceiling total 10.00 EUR (0.00 left); monthly not set"
);
}
/// **No bearer, no server** (stinky #9): `serve` with an empty bearer
/// refuses at once, with an ALERT that names the token file. It never
/// serves OPEN by accident; a dev server asks for that by name.
#[test]
fn an_empty_bearer_is_refused_by_name_and_never_served_open() {
let dir = std::env::temp_dir().join(format!("monetize-no-bearer-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let token_file = dir.join("token");
let path = dir.join("monetize.toml");
let mut cfg = crate::config::Config {
data_dir: dir.join("data"),
token_file: Some(token_file.clone()),
..Default::default()
};
cfg.grpc_addr = "127.0.0.1:0".parse().unwrap();
for bearer in ["", " "] {
let (tx, rx) = std::sync::mpsc::channel();
let (cfg, path, bearer) = (cfg.clone(), path.clone(), bearer.to_string());
std::thread::spawn(move || {
let r = super::serve(cfg, &path, bearer, &crate::registry::Linked::open());
let _ = tx.send(r.map_err(|e| format!("{e:#}")));
});
let r = rx
.recv_timeout(std::time::Duration::from_secs(20))
.expect("an empty bearer is refused at once, not served");
let e = r.expect_err("no bearer: not served");
assert!(
e.contains("ALERT") && e.contains(&token_file.display().to_string()),
"{e}"
);
}
let _ = std::fs::remove_dir_all(&dir);
}
}