monetize_product/

lib.rs

//! **`Product` — the plugin that makes monetize speak one product's language.**
//!
//! monetize handles MANY products, and each product's plugin lives in this
//! repository. So monetize is
//! one service for gunnar.rs, holger.rs, njord, …, and each product is a crate under
//! `products/<name>` implementing this trait. The product itself carries only the thin
//! `monetize-embed` hook behind a cargo feature; it never learns what capacity costs.
//!
//! ```text
//!   product box ──Usage (gRPC, the product's own API)──▶ Product plugin ──▶ monetize core
//!   product box ◀──Entitlement (product's own API)──── Product plugin ◀── monetize core
//! ```
//!
//! # Two directions, both facts
//!
//! * [`Product::read_usage`] — what a tenant is consuming, in the product's own units
//!   (gunnar: pack bytes, read-cache bytes, LFS bytes, tombstoned bytes, open-store RAM;
//!   holger: whatever holger meters). The plugin maps them to [`Usage`], a flat bag of
//!   named meters, so core never has product-specific fields.
//! * [`Product::push_entitlement`] — the verdict, written back over the product's own
//!   control plane (gunnar: `Entitlement.Set` with `source = payment:<vendor>:<ref>`).
//!   Signed by monetize's key; the product appends it to its attestation log.
//!
//! # A plugin never blocks the product
//!
//! Counting is done **inside the product, always, open source** — a self-hoster wants the
//! numbers too. The plugin only *reads* them. If monetize is down the product keeps
//! serving on its cached entitlement; the plugin's job on reconnect is to catch up, not
//! to have been in the request path.

use std::collections::BTreeMap;

/// **Reading a daily `*_today` counter out of a series of readings** — the fold
/// behind the console's SILENT and FRICTION lists. Pure; see the module doc for
/// why summing the readings is the wrong arithmetic and why it lives here.
pub mod flow;

/// A tenant, as the product names it. gunnar: the `Namespace` name (`team/sub`).
#[derive(Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Debug, serde::Serialize, serde::Deserialize)]
pub struct TenantId(pub String);

/// **Whether the product will actually POLICE a cap on this meter.**
///
/// CPU and RAM exist as quota even where gunnar cannot cap them, and that gap is
/// narrowed without redoing the engine.
///
/// An order may cap any declared meter. But a cap on a meter nothing checks is a
/// promise nobody keeps, and the tenant cannot tell the difference from the
/// invoice. So the product says, per meter, which kind of promise it is — and
/// the console shows it beside the number rather than letting every cap look
/// like a guarantee.
///
/// This is the smallest honest way to let CPU and RAM be sold: they ARE bought,
/// with real money, and they DO change what the box can do — they are simply not
/// something a git server polices per namespace. Saying so is better than either
/// refusing to sell them or pretending they are enforced.
#[derive(Clone, Copy, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub enum Enforcement {
    /// **The product refuses when over.** A cap here is a wall the tenant hits.
    /// gunnar: `pack_bytes` (receive-pack refused before the bytes),
    /// `cache_bytes` (the explode budget).
    Enforced,
    /// **Counted and reported, not policed.** The number is true and monetize
    /// can bill on it or a human can act on it, but nothing refuses. A cap here
    /// is a threshold, not a wall.
    Measured,
    /// **Neither counted nor policed BY THE PRODUCT — it is iron monetize
    /// bought.** CPU and RAM: a bigger plan at the provider, real money, real
    /// effect, and no per-namespace check anywhere in the product. The
    /// provider's invoice is the enforcement.
    Provisioned,
}

impl Enforcement {
    /// One line for the console, beside the cap.
    pub fn describe(self) -> &'static str {
        match self {
            Enforcement::Enforced => "the product refuses when over",
            Enforcement::Measured => "counted and reported; nothing refuses",
            Enforcement::Provisioned => "provisioned at the cloud provider; not policed by the product",
        }
    }

    /// Is a cap on this meter a wall the tenant will actually hit?
    pub fn is_a_wall(self) -> bool {
        matches!(self, Enforcement::Enforced)
    }
}

/// **What a cap on this meter is MADE OF**, for the meters where a cap is a
/// promise of CAPACITY rather than a count — and therefore what UNIT the price
/// list prices it in and what iron the transaction buys for it.
///
/// A plan that sells bytes it never buys is the one bug here that silently
/// converts money into nothing. Since the plan catalogue went (2026-09-05), an
/// ORDER names caps and nothing else, and this is the field that turns a cap
/// into a purchase: `monetize::order::resources_for` translates the DELTA on a
/// disk-backed meter into a `Resource::Disk`, and `monetize::PriceList` prices
/// each backing's delta by its own unit.
///
/// It lives on the meter and not on the order because only the PRODUCT knows
/// what its meter is made of — that `pack_bytes` is block storage and `pushes`
/// is a count of events that costs nothing to promise.
#[derive(Clone, Copy, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub enum Backing {
    /// **Bytes of block storage.** A delta here becomes a `Resource::Disk` of
    /// `ceil(bytes / 2^30)` GiB, and is priced per GiB per month
    /// (`disk_gib_month` on the price list).
    DiskBytes,
    /// **CPU, in millicores (1000 = one core).** Priced per core per month
    /// (`cpu_core_month`). monetize does NOT translate millicores into a
    /// provider's plan name (`2xCPU-4GB` is a name, and a table of names goes
    /// stale every time a provider renames a size), so a delta here buys no
    /// cloud resource of its own: the ceiling is the product's own measurement
    /// (`Product::can_absorb`) and the box's declared limit
    /// (`CloudProvider::ceilings`).
    CpuMillicores,
    /// **RAM, in bytes.** Priced per GiB per month (`ram_gib_month`); the same
    /// ceiling rule as [`Backing::CpuMillicores`].
    RamBytes,
}

impl Backing {
    /// The price-list key this backing is priced under, and the unit it means:
    /// `disk_gib_month` | `cpu_core_month` | `ram_gib_month`.
    pub fn unit(self) -> &'static str {
        match self {
            Backing::DiskBytes => "disk_gib_month",
            Backing::CpuMillicores => "cpu_core_month",
            Backing::RamBytes => "ram_gib_month",
        }
    }

    /// Every backing, in the order the price list is written in.
    pub const ALL: [Backing; 3] = [Backing::DiskBytes, Backing::CpuMillicores, Backing::RamBytes];

    /// The backing a price-list key names, if any.
    pub fn from_unit(unit: &str) -> Option<Backing> {
        Backing::ALL.into_iter().find(|b| b.unit() == unit)
    }

    /// **Price `delta` of this backing at `minor_per_unit_month`, for one
    /// month**, in minor units. Integer arithmetic, rounded UP to the unit for
    /// bytes (a customer who asks for one byte over a GiB is sold the next GiB,
    /// which is what the provider sells us) and exact per millicore for CPU
    /// (`minor × millicores / 1000`).
    pub fn price_month(self, delta: u64, minor_per_unit_month: u64) -> u64 {
        const GIB: u64 = 1 << 30;
        match self {
            Backing::DiskBytes | Backing::RamBytes => delta.div_ceil(GIB).saturating_mul(minor_per_unit_month),
            Backing::CpuMillicores => u64::try_from(u128::from(delta) * u128::from(minor_per_unit_month) / 1000).unwrap_or(u64::MAX),
        }
    }
}

/// **WHERE a meter's number is even meaningful: per tenant, or per product.**
///
/// Disk is the whole game, and it is per tenant. CPU and RAM stay in the model,
/// but for gunnar they are not interesting per user — they are interesting per
/// product: a tenant does not own cores, the box does.
///
/// Orthogonal to [`Enforcement`], and the pair is what makes a fleet total
/// honest. `Enforcement` says whether a cap is a WALL. Scope says whether adding
/// one tenant's figure to another's produces anything at all:
///
/// * `pack_bytes` is per tenant, so nine tenants' pack bytes sum to the fleet's
///   pack bytes;
/// * `cpu_millicores` is per product, so nine tenants' CPU does not sum to
///   anything — nobody measured a tenant's cores, and adding nine numbers that
///   were never measured is arithmetic on nothing.
///
/// **It is the PRODUCT's call, not a global rule.** gunnar meters disk per tenant
/// and CPU per product; another product may legitimately meter CPU per user, or
/// disk only in total. [`Product::meters`] is already that seam, and nothing
/// outside a product decides which of its meters are per-tenant. `monetize::fleet`
/// sums only [`Scope::Tenant`], whatever a product declares.
#[derive(Clone, Copy, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub enum Scope {
    /// **Measured, or sold, per TENANT.** A fleet figure is the sum over tenants.
    /// The default, because it is what a meter usually is.
    Tenant,
    /// **Measured, or bought, for the WHOLE PRODUCT.** It appears ONCE, on the
    /// product screen, and never in a per-tenant sum. A cap on it may still be
    /// sold — CPU and RAM are bought with real money — it simply is not a
    /// quantity any one tenant holds.
    Product,
}

impl Scope {
    /// May a figure for this meter be summed across tenants? Only [`Scope::Tenant`].
    pub fn sums_across_tenants(self) -> bool {
        matches!(self, Scope::Tenant)
    }

    /// `tenant` | `product` — the wire and JSON spelling, one writer.
    pub fn name(self) -> &'static str {
        match self {
            Scope::Tenant => "tenant",
            Scope::Product => "product",
        }
    }
}

/// **One meter a product declares**: its name, what it means, and whether a cap
/// on it is policed. An order may only cap a declared meter, and the console
/// shows [`Enforcement`] beside the number.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct Meter {
    /// The wire name, and the key in [`EntitlementFact::caps`] and [`Usage::meters`].
    pub name: &'static str,
    /// One line, the plugin's own words.
    pub meaning: &'static str,
    pub enforcement: Enforcement,
    /// **Does [`Product::read_usage`] carry a reading for this meter?**
    ///
    /// A meter exists to do one or both of two jobs: carry a usage READING, and
    /// accept a CAP. Most do both. Two kinds do not, and conflating them was
    /// caught by a test rather than by thinking:
    ///
    /// * `cpu_millicores`, `ram_bytes` — the product has no idea what was
    ///   bought; monetize does, from the order. Reporting `0` would read as "you
    ///   have none", which is worse than saying nothing.
    /// * `concurrent_transfers` — a cap the engine enforces instantly, with no
    ///   stored reading to bill from. The limit is real; the gauge does not exist.
    ///
    /// So a plugin's `read_usage` must report EXACTLY the meters with this set,
    /// which is an equality a test can hold in both directions: an undeclared
    /// reading is a leak, and a declared reading that stopped arriving is a
    /// silently emptied bill.
    pub reports_usage: bool,
    /// **What a cap on this meter is made of**, or `None` when a cap here costs
    /// nothing to grant (a count of pushes, a number of repositories the
    /// existing box already holds). `Some` is what makes an order on this meter
    /// a PURCHASE: the delta is priced by the backing's unit and, for disk,
    /// bought as iron. See [`Backing`].
    pub backing: Option<Backing>,
    /// **Whether this meter's number is per tenant or per product.** See
    /// [`Scope`]; it is what keeps a fleet total from adding up figures nobody
    /// measured per tenant.
    pub scope: Scope,
}

impl Meter {
    /// A meter that both reports a reading and may be capped — the common case.
    pub const fn new(name: &'static str, meaning: &'static str, enforcement: Enforcement) -> Meter {
        Meter { name, meaning, enforcement, reports_usage: true, backing: None, scope: Scope::Tenant }
    }

    /// A meter that may be CAPPED but carries no reading. See
    /// [`Meter::reports_usage`] for the two kinds and why `0` is not a truthful
    /// substitute.
    pub const fn cap_only(name: &'static str, meaning: &'static str, enforcement: Enforcement) -> Meter {
        Meter { name, meaning, enforcement, reports_usage: false, backing: None, scope: Scope::Tenant }
    }

    /// The same meter, declaring what a cap on it is MADE OF — and so what an
    /// order on it buys and what the price list prices it by. A
    /// [`Enforcement::Provisioned`] meter without one is a cap monetize can
    /// neither price nor buy; gunnar's own test holds its list to that.
    pub const fn with_backing(self, backing: Backing) -> Meter {
        Meter { backing: Some(backing), ..self }
    }

    /// The same meter, measured for the WHOLE PRODUCT rather than per tenant —
    /// so a fleet total leaves it out instead of adding up figures nobody took.
    /// See [`Scope`].
    pub const fn per_product(self) -> Meter {
        Meter { scope: Scope::Product, ..self }
    }
}

/// **What a product says it could still serve, or why it could not say.**
///
/// The other half of the oversell number: `sum(what has been sold) − servable`.
/// Without it monetize could sell ten tenants 10 GiB each on a box with 55 GiB
/// servable and nothing would object until the sixth push.
///
/// The two arms are the distinction `BASE-MODEL.md` rule 3 turns on and the same
/// one [`Product::can_absorb`] already makes: **an unanswered capacity question
/// is a measurement that did not happen, not a full disk and not an empty one.**
/// The live gunnar.rs appliance predates gunnar's `Admin.Capacity` RPC and
/// answers [`Servable::Unmeasured`] today; a fleet total that turned that into a
/// zero would report every deployment as catastrophically oversold, and one that
/// turned it into infinity would report every deployment as fine. Neither is a
/// measurement.
/// **Can this product's store grow while it runs?** Carried on
/// [`Servable::Measured`] beside the bytes, because the two answer different
/// questions and an operator acts differently on each: a full box that can grow
/// wants a disk; a full box that is SEALED wants its set enlarged by an operator
/// maintenance operation (the volumes it already has, grown in place) or a fresh
/// install onto a larger one, and a disk bought for it is billed and invisible.
///
/// Each arm carries the PRODUCT's own remediation sentence (gunnar's
/// `remediation()`), so the console renders what the product said rather than a
/// paraphrase of it — `UI.md` screen 2's `⚠ cannot grow` badge holds exactly
/// that paragraph.
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub enum Growth {
    /// Cannot gain capacity while it runs, and nothing on the machine can give
    /// it any. `GROWTH_SEALED` on gunnar's wire.
    Sealed(String),
    /// The product will not grow the set itself, but an operator can, out of
    /// band. `GROWTH_OPERATOR_ONLY`.
    OperatorOnly(String),
    /// The product could not tell. NOT a promise that it can grow.
    Unknown(String),
    /// **Grows at its next restart** once every member volume has been
    /// enlarged to one equal size (gunnar's `GROWTH_AT_RESTART`, 2026-09-14).
    /// Still no hot-plug: a disk ATTACHED to it is as invisible as under
    /// `Sealed`; what helps is the outside actor enlarging the members
    /// ([`ApplianceGrow`] + `monetize_cloud::grow`).
    AtRestart(String),
    /// **Grows without a restart**: the product drains, unmounts and grows into
    /// enlarged members while its control plane stays up (gunnar's
    /// `GROWTH_AT_RUNTIME`). The shape [`ApplianceGrow`] drives.
    AtRuntime(String),
}

impl Growth {
    /// The wire word: `sealed` | `operator_only` | `unknown` | `at_restart` |
    /// `at_runtime`.
    pub fn name(&self) -> &'static str {
        match self {
            Growth::Sealed(_) => "sealed",
            Growth::OperatorOnly(_) => "operator_only",
            Growth::Unknown(_) => "unknown",
            Growth::AtRestart(_) => "at_restart",
            Growth::AtRuntime(_) => "at_runtime",
        }
    }

    /// The product's own sentence about what to do.
    pub fn detail(&self) -> &str {
        match self {
            Growth::Sealed(s) | Growth::OperatorOnly(s) | Growth::Unknown(s) | Growth::AtRestart(s) | Growth::AtRuntime(s) => s,
        }
    }
}

#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub enum Servable {
    /// The product measured its own store.
    Measured {
        /// **Bytes that may still be handed to tenants**, beyond what they already
        /// hold, after whatever reserve the product keeps for itself. gunnar's
        /// `servable_bytes` (`free − disk floor`) — never its `free_bytes`, which
        /// includes a reserve monetize must not sell.
        servable_bytes: u64,
        /// The whole store, reserve and all. `used + servable_bytes` is what the
        /// fleet could grow to; `total_bytes` is larger than that by the reserve
        /// and by anything on the filesystem that is not tenant data. 0 when the
        /// product measures what is left but not what there is in total.
        total_bytes: u64,
        /// **Whether the store can EVER hold more than `total_bytes`.** A
        /// measurement of the box, independent of the bytes: gunnar's
        /// `Admin.Capacity` answers it as `growth`, and it is the field that
        /// stops "buy another disk" being the reflex answer to a full fleet.
        growth: Growth,
        /// **The box's MEASURED ceilings on product-scope meters**, by the
        /// product's own meter name: gunnar's `cpu_millicores_total` and
        /// `ram_bytes_total` (Admin.Capacity, since gunnar 744bf72d) land here
        /// as `cpu_millicores` / `ram_bytes`. A meter ABSENT here was not
        /// measured — the wire spells that as an absent field, never as 0, and
        /// a reader must never take a 0 as a ceiling. An order that would
        /// raise a product-scope cap past a number here is refused by name;
        /// the box's DECLARED ceilings (`CloudProvider::ceilings`) carry only
        /// for a meter this map does not hold.
        ceilings: BTreeMap<String, u64>,
    },
    /// **It could not be asked, or would not say — which is not zero.** The
    /// string names WHICH: an RPC the deployed build predates, a control plane
    /// that is down, a plugin that was never wired.
    Unmeasured(String),
}

/// One named meter. Names are the product's, documented in its plugin crate, e.g.
/// `pack_bytes`, `cache_bytes`, `lfs_bytes`, `tombstoned_bytes`, `open_store_ram_bytes`,
/// `pushes`, `anonymous_reads`. Values are the product's units, usually bytes or counts.
#[derive(Clone, PartialEq, Eq, Debug, Default, serde::Serialize, serde::Deserialize)]
pub struct Usage {
    pub meters: BTreeMap<String, u64>,
    pub measured_at_unix_ms: u64,
}

/// **One tenant's FLOW for one UTC day** — what MOVED, as against what is held.
///
/// Every number in [`Usage`] is a STOCK reading: it answers *how much is on the
/// product's disk*, and it reads exactly the same for a customer who worked
/// this morning and one who last touched the product in March. monetize could
/// therefore bill nine tenants correctly and not know that six of them had
/// stopped. This is the other kind, and it is what the SILENT and FRICTION
/// lists are made of.
///
/// **The meters are the product's own names**, exactly as [`Usage::meters`] is,
/// and for the same reason: core has never held a product-specific field and
/// must not gain one here. gunnar fills `bytes_in`, `bytes_out`, `pushes`,
/// `fetches`, `lfs_bytes_in`, `lfs_bytes_out`, `refused`, its three
/// `refused_*` reasons, `auth_failures`, `distinct_principals` and the two
/// `last_*_unix_ms` timestamps. Another product will fill something else, and
/// nothing here needs to know.
///
/// **A day with no traffic has NO ROW.** Not a row of zeroes: the caller knows
/// which days it asked for, and zero-filling would put a bar of height zero
/// beside a bar that means *not counted*.
#[derive(Clone, PartialEq, Eq, Debug, Default, serde::Serialize, serde::Deserialize)]
pub struct ActivityDay {
    /// Midnight UTC of the day, in Unix milliseconds.
    pub day_unix_ms: u64,
    /// The product's own meter names. See the type's doc.
    pub meters: BTreeMap<String, u64>,
}

/// **A tenant's kept activity, and how far back the product remembers.**
///
/// The second half is not decoration. An empty `days` means one of two opposite
/// things — *nothing happened* or *you asked about a time the product no longer
/// remembers* — and only `retained_days` can tell them apart. A console that
/// drew an empty series as "silent" without it would report every tenant on a
/// freshly installed box as a lapsed customer.
#[derive(Clone, PartialEq, Eq, Debug, Default, serde::Serialize, serde::Deserialize)]
pub struct Activity {
    /// Oldest first. Days with no traffic are absent.
    pub days: Vec<ActivityDay>,
    /// How many days the product keeps. `0` when it would not say.
    pub retained_days: u64,
}

/// The verdict monetize pushes back. Product-agnostic; the plugin maps it to the
/// product's own enum (gunnar: `EntitlementState`).
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub struct EntitlementFact {
    pub tenant: TenantId,
    /// **The ORDER this verdict came from** — the ledger reference
    /// (`<product>/<tenant>/<date>+<caps>`) for a purchase, an operator's own
    /// label (or nothing) for a hand-set fact. The field is still called `plan`
    /// on the wire and in the signed canonical form (`monetize_embed::signing`),
    /// because gunnar verifies that form and a renamed field would invalidate
    /// every signature a deployed gunnar checks; the CONTENT is an order
    /// reference since the plan catalogue went on 2026-09-05.
    pub plan: String,
    pub state: State,
    pub paid_until_unix_ms: Option<u64>,
    /// Per-meter caps the product enforces itself (gunnar: `pack_quota_bytes`,
    /// `explode_budget_bytes`). Absent = product default.
    pub caps: BTreeMap<String, u64>,
    /// `operator` | `payment:<vendor>:<reference>` — lands in the product's attestation log.
    pub source: String,
    /// Ed25519 over the canonical JSON of the fields above, by monetize-server's key.
    /// **The V1 form** — it does NOT cover [`EntitlementFact::issued_unix_ms`], and
    /// that is deliberate: it is the signature an appliance built before
    /// 2026-09-17 computes, and it must keep verifying there for ever. See
    /// [`Self::issued_signature`].
    pub signature: Vec<u8>,
    /// **When monetize issued this verdict — the field that makes a fact good
    /// ONCE.**
    ///
    /// Without it an `EntitlementFact` is replayable for ever: capture a `Paid`
    /// fact, wait for the tenant to lapse, push the captured bytes back, and the
    /// signature still verifies because it is a real signature. `paid_until`
    /// cannot tell the two apart — a lapse keeps the date and moves the ladder —
    /// so the only thing that can is something MONOTONIC inside the signed form.
    /// A product refuses a fact that is not newer than the one it holds
    /// (gunnar: `EntitlementSlot::set`).
    ///
    /// Unix milliseconds, and not a sequence number, for the same reason
    /// [`crate`]'s sibling [`monetize_embed::signing::Snapshot`] chose one: a
    /// clock needs no durable per-tenant counter on monetize's side, so a
    /// restored ledger cannot rewind one and mint facts every appliance in the
    /// field then refuses for ever. One idea in this system, not two.
    ///
    /// `None` is a fact signed before this field existed. It stays legal, and a
    /// product accepts it — until that product has seen ONE stamped fact for the
    /// tenant, after which the unstamped form is a downgrade and is refused.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub issued_unix_ms: Option<u64>,
    /// **The V2 signature**: Ed25519 over the canonical JSON of every field
    /// above INCLUDING `issued_unix_ms`
    /// (`monetize_embed::signing::fact_message_issued`).
    ///
    /// Two signatures and not one, because a fact has to be readable by two
    /// generations of appliance at once. An appliance that predates this field
    /// reads only fields 1-7 off the wire, computes the V1 form, and checks
    /// [`Self::signature`] — so it accepts a stamped fact unchanged, and a
    /// paying customer on an un-upgraded box loses nothing. An appliance that
    /// knows the field checks BOTH, so the issue time is signed and cannot be
    /// added, moved or bumped by whoever relays the fact.
    ///
    /// Empty exactly when `issued_unix_ms` is `None`; neither is legal without
    /// the other.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub issued_signature: Vec<u8>,
}

/// The ladder. Numbers (grace/retention days) are the deployment's policy
/// (`monetize::Policy`), not the enum's.
#[derive(Clone, Copy, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub enum State {
    Free,
    Paid,
    Grace,
    Suspended,
    Retention,
}

#[derive(Clone, Debug)]
pub enum ProductError {
    /// The product's control plane refused (auth, unknown tenant).
    Refused(String),
    /// Product unreachable. Retryable; nothing was written.
    Unavailable(String),
}

impl std::fmt::Display for ProductError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ProductError::Refused(r) => write!(f, "product refused: {r}"),
            ProductError::Unavailable(r) => write!(f, "product unavailable: {r}"),
        }
    }
}
impl std::error::Error for ProductError {}

/// **Where the room for a target comes from, as the product measured it** —
/// what [`Product::can_absorb`] answers when it can serve.
///
/// A yes is two different facts, and the difference is money. gunnar's
/// `Admin.Capacity` answers a `verdict` and a `growth`, and the three readings
/// of them a seller can act on are: there is room already (buy nothing), there
/// is no room but a bought disk will be used (buy, and somebody has to extend a
/// filesystem), and there is no room and a bought disk will NOT be used (refuse).
/// The third is a refusal and never reaches this type. Before this type existed
/// a yes was `Ok(())` and the cloud was asked to `ensure` a disk regardless. With
/// the front's configuration (`pool_reserve_gib` covering the whole data set)
/// UpCloud's pool arithmetic answers "short" for every order, so every yes
/// bought and attached a disk to an appliance that had just said it had room —
/// read from the code and the config, not from a bill.
///
/// The transaction turns it into `monetize_cloud::Room` for the provider, so
/// the cloud is told what the product measured and never has to know which
/// product that was.
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub enum Absorb {
    /// **Served out of what the product already has.** Nothing may be bought
    /// for it: a disk bought on top of this answer is billed and not needed.
    OnHand,
    /// **Short today, and bought iron becomes usable** — by the hand `note`
    /// names, in the product's own words (gunnar: an operator extends the
    /// filesystem out of band, and gunnar measures the larger one on its next
    /// look). Carried onto the purchase row so the ledger records that
    /// somebody still has a job to do after the disk is attached.
    WithIron { note: String },
}

/// **What a metered verb answered** — see [`Product::metered_door`]. Three
/// arms, because "nobody could ask" is neither a paid build nor a community
/// one, and a deployment that sells against any of the last two is selling
/// something it cannot deliver.
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub enum MeteredDoor {
    /// **The box answered a metered verb.** It has the paid services compiled
    /// in; the string is whatever the plugin wants to record about the proof
    /// (the verb it called, the build it saw).
    Enforcing(String),
    /// **The box answered `UNIMPLEMENTED`.** A community build. Every paid
    /// signup against it will take the money and fail to grant the
    /// entitlement, so a deployment that sells is misconfigured, loudly.
    Absent(String),
    /// **Nobody could ask** — the box is down, the plugin has no probe, nothing
    /// has run yet. NOT a yes and NOT a no; carries which.
    Unproven(String),
}

impl MeteredDoor {
    /// `enforcing` | `absent` | `unproven` — the wire, JSON and log spelling,
    /// one writer.
    pub fn name(&self) -> &'static str {
        match self {
            MeteredDoor::Enforcing(_) => "enforcing",
            MeteredDoor::Absent(_) => "absent",
            MeteredDoor::Unproven(_) => "unproven",
        }
    }
    /// The plugin's own words about the answer.
    pub fn detail(&self) -> &str {
        match self {
            MeteredDoor::Enforcing(s) | MeteredDoor::Absent(s) | MeteredDoor::Unproven(s) => s,
        }
    }
    /// **Has the paid door been PROVEN to be there?** False for `Unproven`,
    /// which is the whole point of the arm: a deployment may not call itself
    /// healthy on the strength of a question nobody answered.
    pub fn proven(&self) -> bool {
        matches!(self, MeteredDoor::Enforcing(_))
    }
}

/// The plugin seam. Sync for the same reason as the vendor traits.
pub trait Product: Send + Sync {
    /// `gunnar`, `holger`, `njord`. Also the first segment of every payment reference.
    fn id(&self) -> &'static str;
    /// The meters this product declares — the UI and the order form read this,
    /// so an order can only cap a meter that exists, and the console can say
    /// whether a cap on it is policed ([`Enforcement`]).
    fn meters(&self) -> &[Meter];

    fn list_tenants(&self) -> Result<Vec<TenantId>, ProductError>;
    fn read_usage(&self, tenant: &TenantId) -> Result<Usage, ProductError>;

    /// **What the tenant has been DOING, day by day** — the flow series behind
    /// the console's SILENT and FRICTION lists.
    ///
    /// [`Product::read_usage`] answers *how much is held*. Only this answers
    /// *is this customer using the product, has it gone quiet, and is it being
    /// refused* — and no stock meter has ever been able to.
    ///
    /// The window is Unix milliseconds, inclusive, resolved by the PRODUCT to
    /// whole UTC days: `0` for `since` means the whole kept window, `0` for
    /// `until` means now. The product is the half that knows what it keeps, and
    /// a second arithmetic here would be a second answer to one question.
    ///
    /// # The default REFUSES BY NAME
    ///
    /// [`Product::purge_tenant`]'s rule, for the same shape of danger. The
    /// permissive answer here is an empty series, and an empty series is
    /// rendered as *this tenant has gone quiet* — so a product that never
    /// implemented this would put every one of its customers on the SILENT list
    /// and an operator would go and ask them why they had stopped. A refusal
    /// naming the product is the safe wrong answer, and it says which plugin
    /// owes the work.
    fn read_activity(
        &self,
        tenant: &TenantId,
        since_unix_ms: u64,
        until_unix_ms: u64,
    ) -> Result<Activity, ProductError> {
        let _ = (tenant, since_unix_ms, until_unix_ms);
        Err(ProductError::Refused(format!(
            "the {} plugin cannot report a tenant's activity: it has no flow series, so an empty              answer here would read as a customer who had gone quiet rather than as a question              nobody asked the product.",
            self.id()
        )))
    }
    fn push_entitlement(&self, fact: &EntitlementFact) -> Result<(), ProductError>;

    /// **Can this product actually DELIVER these caps to this tenant, today —
    /// and out of what?** `Ok` says where the room comes from ([`Absorb`]);
    /// `Err(reason)` is a refusal that names what is missing, in words for an
    /// OPERATOR (the transaction gives a customer its own sentence instead).
    ///
    /// # Why it exists
    ///
    /// Everything else in monetize checks whether the money can be taken and
    /// whether the iron can be bought. Nothing asked the third question, and it
    /// is the one that decides whether the sale is honest: **once the disk is
    /// attached, can the product use it?**
    ///
    /// For gunnar today the answer is often NO. The appliance's `/data` is a
    /// 4x75 GB raid0 xfs set laid once at install (`InstallMode::Fresh`), there
    /// is no runtime data-set growth, and gunnar is PID 1 with no shell — so a
    /// disk monetize buys and attaches is **billed and invisible**. Selling a
    /// bigger quota there does not fail; it succeeds, charges the customer, signs
    /// a fact promising capacity, and the capacity is not there. That is the
    /// worst failure shape this system has: money moved, everything green, the
    /// promise hollow.
    ///
    /// # The contract
    ///
    /// * `caps` is the FULL target cap set, not a delta — the product is asked
    ///   about the world it would have to serve, not about the change.
    /// * It is asked **before the reserve**, so a refusal costs nothing and
    ///   nothing has to be unwound. See
    ///   `monetize::transaction::Transaction::increase`.
    /// * `Ok` is a claim, not a shrug. A plugin that cannot tell must say so
    ///   in an `Err`, because a plugin that guesses yes is indistinguishable
    ///   from one that knows, right up until a tenant is charged for nothing.
    /// * `Ok(Absorb::OnHand)` forbids a purchase and `Ok(Absorb::WithIron)`
    ///   permits one. A product whose store cannot use bought iron must never
    ///   answer `WithIron`: it refuses instead, and says what would help.
    /// * It has **no default implementation**, deliberately. A default `Ok(())`
    ///   would let a product that never thought about this answer yes forever,
    ///   which is exactly the silence this method exists to break.
    ///
    /// It may talk to the product (it is the plugin's own control plane), so it
    /// may fail for the usual reasons; report those as a refusal with the reason
    /// in it rather than inventing a yes.
    fn can_absorb(&self, tenant: &TenantId, caps: &BTreeMap<String, u64>) -> Result<Absorb, String>;

    /// **How much this product could still serve, across the whole product.**
    ///
    /// The denominator of `UI.md`'s `184 / 300 GiB` and the servable half of the
    /// oversell number (`monetize::fleet`). [`Product::can_absorb`] asks the same
    /// box a narrower question — *can you take THIS tenant to THIS cap* — and
    /// answers yes or no; this one asks for the figure, because a fleet total
    /// needs a number and not a verdict.
    ///
    /// **It has no default implementation, for [`Product::can_absorb`]'s reason.**
    /// A default would let a product that never thought about capacity answer
    /// forever, and whichever constant it returned would be a lie: 0 reads as a
    /// full disk, `u64::MAX` reads as an empty one, and the truth for a product
    /// that cannot measure is [`Servable::Unmeasured`] — which the plugin must
    /// say in its own words, naming what is missing.
    ///
    /// It may talk to the product, so it may fail for the usual reasons; report
    /// those as `Unmeasured` with the reason in them rather than inventing a
    /// number.
    fn servable(&self) -> Servable;

    /// **Is the PAID DOOR actually there?** One call to a verb only a metered
    /// build answers, so a deployment that sells can prove the box it sells
    /// against can enforce what it sold.
    ///
    /// # Why this is not covered by anything else
    ///
    /// Everything the poller does — `list_tenants`, `read_usage` — is an OPEN
    /// build's service. gunnar's community ISO answers both perfectly, and
    /// `poll tenants=11 failures=0` therefore proves connectivity and nothing
    /// whatever about entitlements. The first thing that finds out is a paying
    /// customer: the sink creates the principal, pushes the entitlement,
    /// `Entitlement.Set` answers `UNIMPLEMENTED`, and the signup is refused
    /// with the money already taken (`monetize_server::signup::Owed`). The last
    /// proof this estate's appliance is a metered build is a journal line from
    /// 2026-09-15.
    ///
    /// So: a READ-ONLY metered verb, asked on a schedule. It must not create,
    /// change or delete anything — it is called on every poll tick, against a
    /// live box, forever.
    ///
    /// # The default is UNPROVEN, and unproven is not "yes"
    ///
    /// Unlike [`Product::can_absorb`] and [`Product::servable`], which have no
    /// default because every constant they could return is a lie about
    /// capacity, this one has a safe answer that is not a claim:
    /// [`MeteredDoor::Unproven`], naming the plugin that owes the work. A
    /// default of `Enforcing` would let every plugin that never thought about
    /// it assert the one thing this exists to check.
    fn metered_door(&self) -> MeteredDoor {
        MeteredDoor::Unproven(format!(
            "the {} plugin has no metered-door probe, so nothing has asked this box whether it can enforce an entitlement at all",
            self.id()
        ))
    }

    /// **Delete everything this tenant has in the product, and keep them locked
    /// out while it happens.**
    ///
    /// The product half of a super purge. `monetize` can give the IRON back on
    /// its own — `transaction::release_tenant` and
    /// `CloudProvider::release_for` — and it cannot delete a byte of what is on
    /// it, because only the product knows what a tenant's data IS. In gunnar it
    /// is every repository in the account plus two classes of bytes that are
    /// not in the catalog at all; nothing on this side of the seam could
    /// enumerate that.
    ///
    /// # It must LOCK, and the lock is the product's
    ///
    /// A purge is not atomic — it walks and reclaims over minutes while the
    /// product goes on serving — so anything the tenant does during the walk
    /// lands behind it. The lockout that prevents that can only live where the
    /// requests arrive, which is the product. gunnar's is `Accounts.Gate`, and
    /// it is total rather than read-only because in gunnar a read creates
    /// things: a fetch explodes objects into a cache the purge just unlinked.
    ///
    /// [`Purged::still_locked`] carries the outcome back, because a product
    /// that emptied a tenant and could not let them back in has left an account
    /// nobody can use, and that must not read as success anywhere above here.
    ///
    /// # The default REFUSES BY NAME
    ///
    /// Unlike [`Product::can_absorb`] and [`Product::servable`], which have no
    /// default at all because every constant they could return is a lie about
    /// capacity, this one's dangerous answer is a permissive `Ok` — a product
    /// that never implemented it reporting a tenant's data gone when it is
    /// still there, after which a super purge would cheerfully destroy the
    /// disks it is on. A refusal naming the product is the safe wrong answer
    /// and it says which product owes the work.
    fn purge_tenant(&self, tenant: &TenantId, reason: &str) -> Result<Purged, ProductError> {
        let _ = (tenant, reason);
        Err(ProductError::Refused(format!(
            "the {} plugin cannot purge a tenant's data: it has no purge verb, so nothing here \
             can promise the tenant's bytes are gone. Empty the tenant in the product itself \
             before releasing its resources.",
            self.id()
        )))
    }

    /// **The product's own half of a data-set growth**, if it has one. `None`
    /// — the default — is a product whose set cannot be grown at runtime from
    /// outside; monetize then refuses a growth by name before touching a
    /// volume. See [`ApplianceGrow`].
    fn grow(&self) -> Option<&dyn ApplianceGrow> {
        None
    }

    /// **The TWIN's own half of ITS data-set growth**, if the product has a twin whose
    /// control plane monetize can reach. Owner ruling 2026-09-22 (T21): the twin, built
    /// from the same medium, grows EXACTLY like main — the same [`ApplianceGrow`] verbs on
    /// the twin's own control plane (drain, unmount, wait, grow, mount; `maintenance`
    /// meanwhile) around the cloud's same `grow_set`, and it is NEVER stopped or
    /// restarted. `None` — the default — is a product whose twin cannot be grown that
    /// way; its twin-growth lane then holds by name rather than stop the box.
    fn twin_grow(&self) -> Option<&dyn ApplianceGrow> {
        None
    }

    /// **The product's own half of a restore from its twin**, if it has one.
    /// `None` — the default — is a product that cannot be re-laid from
    /// outside; a failed growth then parks in `failed` and an operator's
    /// `cargo xtask restore main` is what is left. See [`ApplianceRestore`].
    /// Named `restore_lane`, not `restore`: a product implements BOTH this
    /// trait and [`ApplianceRestore`], whose own verb is `restore`, and two
    /// methods of that name on one type make every call site ambiguous
    /// (E0034). The lane accessor is the one that moves.
    fn restore_lane(&self) -> Option<&dyn ApplianceRestore> {
        None
    }

    /// **The product's half of replacing a dead twin**, if it has one. `None`
    /// — the default — is a product that cannot adopt a standby from
    /// outside; a lost twin then stays lost until an operator acts. See
    /// [`ApplianceTwinship`].
    fn twinship(&self) -> Option<&dyn ApplianceTwinship> {
        None
    }

    /// **The product's twins, as its primary hears them** — one row per twin:
    /// the heartbeat state, and the fill of the twin's data volume the twin
    /// reported on its last poll. What the batcher's twin-fill trigger reads.
    /// The default is no twins: a product without a twin has nothing to grow.
    fn twins(&self) -> Result<Vec<TwinFill>, ProductError> {
        Ok(Vec::new())
    }

    /// **Whether a tenant-presented actor ticket can be verified at all by this
    /// product.** False means a console must not offer self-service renewal —
    /// there is nothing here that could tell a customer from a stranger, so the
    /// route must not exist rather than existing and refusing.
    ///
    /// Reported on the wire as `ProductInfo.tenant_renewal`.
    fn tenant_renewal(&self) -> bool {
        false
    }

    /// **May the bearer of this ticket act for `tenant`? Default DENY.**
    ///
    /// The ticket is the product's own appliance vouching that the human
    /// driving a console holds the tenant they are paying for
    /// (`monetize_embed::ticket`). monetize's gRPC surface authenticates one
    /// shared bearer — an empty one means open — so this is the ONLY thing that
    /// can tell "alice renewing alice" from "somebody renewing alice".
    ///
    /// # Why the argument is raw bytes
    ///
    /// `monetize-embed` — which owns `ActorTicket` and `verify_ticket` —
    /// depends on THIS crate, so this crate cannot name that type without a
    /// dependency cycle. The verification therefore happens in the PLUGIN,
    /// which may depend on `monetize-embed`, and the seam carries the opaque
    /// bytes it was handed. That is not a compromise: the plugin is also the
    /// only half that holds the appliance's key, so it is where the check
    /// belongs whichever way the crates pointed.
    ///
    /// # Why the default refuses
    ///
    /// A plugin is out of tree and nobody edits it when a path like this ships.
    /// A default of `Allowed` would silently turn every such plugin into one
    /// that honours signed bytes it holds no key for; a default that REFUSES
    /// makes an unaudited product safe by omission rather than by somebody
    /// remembering. The cost of the wrong default here is a customer who cannot
    /// renew themselves; the cost of the other one is a stranger who can.
    /// `purpose` and `caps` are what the REQUEST says, and they are arguments
    /// rather than things the plugin reads off the ticket, because a verifier
    /// cannot compare a claim against itself. A ticket names the verb it
    /// authorises and — for an order — the amount it covers; the plugin's job
    /// is to check both against the call that arrived. Passing them in is what
    /// makes "this ticket is for a different order" a refusal that no
    /// implementation can forget to make.
    ///
    /// `caps` is empty for a renewal, and a plugin must refuse a renewal ticket
    /// that carries any.
    fn may_act_for(
        &self,
        _tenant: &TenantId,
        _purpose: &str,
        _caps: &std::collections::BTreeMap<String, u64>,
        _ticket: &[u8],
    ) -> Result<ActorVerdict, ProductError> {
        Ok(ActorVerdict::Refused("this product cannot verify a tenant-presented ticket".into()))
    }
}

/// **What a product says about a ticket somebody presented.**
///
/// Not a `bool` and not a `Result<(), ProductError>`: a refusal is a normal,
/// expected answer that carries a SENTENCE — the words a caller may show — and
/// a `ProductError` means something else entirely (the product's control plane
/// is down, the question could not be asked). Conflating the two would let an
/// unreachable appliance read as a refused ticket, or worse, the reverse.
#[derive(Clone, PartialEq, Eq, Debug)]
pub enum ActorVerdict {
    /// The ticket verifies, names this tenant, this product and this verb, is
    /// inside its life, and has not been spent before.
    Allowed,
    /// Refused, and why — in words a customer surface may show. A verifier must
    /// not name the ticket's own tenant back to whoever asked about another
    /// one; see `monetize_embed::verify_ticket`.
    Refused(String),
}

/// One twin as its primary last heard from it (gunnar's `TwinHeartbeat`).
#[derive(Clone, PartialEq, Eq, Debug, Default, serde::Serialize, serde::Deserialize)]
pub struct TwinFill {
    pub principal: String,
    pub address: String,
    /// `ok` | `behind` | `silent`.
    pub state: String,
    /// Wall clock of the twin's last poll; 0 = never heard.
    pub last_seen_unix_ms: i64,
    pub lag_entries: u64,
    pub disk_total_bytes: u64,
    pub disk_used_bytes: u64,
    /// used × 1000 / total; 0 when total is unknown.
    pub fill_permille: u32,
}

impl TwinFill {
    /// The primary can still hear it: `ok` or `behind`, never `silent` or a word we do not know.
    pub fn visible(&self) -> bool {
        matches!(self.state.as_str(), "ok" | "behind")
    }
}

// ── the appliance's half of a data-set growth ───────────────────────────────

/// **What the appliance does INSIDE the box while the outside actor grows its
/// volumes** (`DATA-SET-GROWTH-FLOW.md` §0 ruling 1, T6): drain, unmount, wait
/// for the members to come back larger, run its engine, mount, reopen. monetize
/// drives it — flush → start → wait `unmounted` → cloud steps → resume → wait
/// `done|failed` — and every verb here is one call on the product's control
/// plane. Vendor-neutral: the words are the phases the flow document names,
/// and the go-ahead is opaque bytes the product verifies on its own terms.
pub trait ApplianceGrow: Send + Sync {
    /// **Prove the twin is caught up** (T11, step 0) — flush replication to
    /// every standby, bounded by `timeout_secs` (0 = the product's own ceiling),
    /// and report. Changes no role and no roster.
    fn flush(&self, timeout_secs: u32) -> Result<FlushReport, ProductError>;
    /// Begin the growth: drain, unmount, wait for the members. `go_ahead` is
    /// the signed approval (see `monetize_embed::signing::go_ahead_message`);
    /// `target_sectors_per_member` is in 512-byte sectors, 0 = "whatever every
    /// member comes back larger at". Answers the status after the call.
    fn start(&self, target_sectors_per_member: u64, go_ahead: &[u8]) -> Result<GrowStatus, ProductError>;
    /// The phase and the members, as the product sees them. **Must answer
    /// while the set is unmounted and, where the product offers it, without a
    /// credential** — the outside actor decides whether to re-attach from it.
    fn status(&self) -> Result<GrowStatus, ProductError>;
    /// "Look again": the members are back. With `give_up`, stop waiting and
    /// serve whatever is there at whatever size it is.
    fn resume(&self, give_up: bool) -> Result<GrowStatus, ProductError>;
}

// ── the appliance's half of a RESTORE from its twin ────────────────────────

/// **What the appliance does when the data set is gone and the twin still has
/// it** — `DATA-SET-GROWTH-FLOW.md` ruling 2: *"If the grow fails, gunnar
/// re-lays (reformats) the data set and pulls EVERYTHING back from the twin.
/// The reformat must never run unless the twin is proven caught up first."*
///
/// Gap **G3** is this verb on the appliance's side; what is here is the seam
/// monetize drives it through, shaped exactly like [`ApplianceGrow`]: one call
/// to start, one to poll, and the phase words are the product's own. monetize
/// owns no second driver — the sequence `cargo xtask restore main` performs by
/// hand (prove the twin → re-lay → replay → verify) is the sequence the
/// product performs here, and the difference is only who said go.
///
/// A product without this lane answers `None` from [`Product::restore`], and a
/// growth that fails then parks in `failed` exactly as it always has.
pub trait ApplianceRestore: Send + Sync {
    /// **Re-lay the set and pull everything back from the twin.** Answers the
    /// status after the call, never when the restore has finished — a replay
    /// is long, and the poll is [`ApplianceRestore::restore_status`].
    ///
    /// `accept_lost` is the operator's acceptance of
    /// `DATA-SET-GROWTH-FLOW.md` §2.1: every ref outside `refs/heads/*` and
    /// `refs/tags/*`, every LFS object, the control database, every
    /// repository's visibility, every tombstone. Without it a restore that
    /// would lose anything REFUSES, and says what.
    fn restore(&self, accept_lost: bool) -> Result<RestoreStatus, ProductError>;
    /// The restore's phase, as the product sees it. Must answer while the set
    /// is unmounted, for the same reason [`ApplianceGrow::status`] must.
    fn restore_status(&self) -> Result<RestoreStatus, ProductError>;
}

/// What a product reports about a restore in progress.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct RestoreStatus {
    /// `idle` | `laying` | `replaying` | `verifying` | `done` | `failed`.
    pub phase: String,
    /// `failed`'s reason; empty otherwise.
    pub why: String,
    /// The twin's head the replay is against.
    pub head_seq: u64,
    /// Entries applied so far.
    pub applied: u64,
    /// **Entries the twin could not give back.** Non-zero is the §2.1 loss,
    /// and a restore started without `accept_lost` refuses rather than
    /// reaching this with a number in it.
    pub lost: u64,
    /// The box needs a restart before it serves again — the ONE place in this
    /// design that restarts the primary.
    pub needs_restart: bool,
    pub since_unix_ms: i64,
    pub detail: String,
}

impl RestoreStatus {
    /// The restore ended, either way.
    pub fn is_terminal(&self) -> bool {
        matches!(self.phase.as_str(), "done" | "failed")
    }
    pub fn is_done(&self) -> bool {
        self.phase == "done"
    }
}

// ── the other direction: the TWIN is the one that died ────────────────

/// **Rebuilding a twin that is gone**, which is the mirror image of
/// [`ApplianceRestore`] and just as necessary: an estate that can recover its
/// primary from its twin but cannot replace a dead twin is an estate that
/// works exactly once.
///
/// Every door this needs already exists on gunnar and none of them is new:
///
/// * `Bootstrap.BecomeTwin{primary}` — *"Make this FRESH box a twin of
///   `primary`. Unauthenticated, once … The box mints its own key, gives up
///   its own log numbers, freezes, and starts following. The answer is the
///   public key to enrol on the primary as a member of group `twin`."*
/// * `Mirror.SetReplication` — *"Add or replace one standby, BY NAME, and
///   start replicating to it now. Idempotent: the same name twice replaces
///   rather than duplicates."*
/// * `Mirror.SnapshotFetch` for a twin that comes up behind a retention hole.
///
/// What was missing is an ACTOR, and for the same reason as on the other
/// side: the twin cannot notice its own death, and the primary has no cloud
/// credential with which to build it a new box. Only the front has both.
///
/// The implementation dials TWO boxes — the fresh candidate's bootstrap door
/// and this primary's mirror door — which is why it is one verb here rather
/// than two: enrolling a key on the primary for a twin that never answered
/// would leave a grant for a box that does not exist.
pub trait ApplianceTwinship: Send + Sync {
    /// **Is there a twin, and is the primary hearing it?** Folded from the
    /// same `Admin.Health` rows [`Product::twins`] serves, so there is one
    /// description of the twin and not two.
    fn twin_health(&self) -> Result<TwinHealth, ProductError>;

    /// **Turn a FRESH box at `twin_addr` into this primary's twin**:
    /// `BecomeTwin` on it, enrol the key it mints on this primary in group
    /// `twin`, then `SetReplication` so it starts following now.
    ///
    /// Refuses by name when the candidate is not fresh — `BecomeTwin` is
    /// *"refused (FAILED_PRECONDITION) unless nothing is enrolled and no
    /// repository exists"* — because making a box fresh is a cloud action and
    /// no product can do it to another box.
    fn adopt_twin(&self, twin_addr: &str) -> Result<TwinAdopted, ProductError>;
}

/// What the primary hears from its standbys right now.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct TwinHealth {
    /// Every standby row, as [`Product::twins`] reports them.
    pub rows: Vec<TwinFill>,
    /// The primary believes it should have a twin — a standby is configured
    /// in its `replication.json`, whether or not it is answering. Absent this,
    /// "no rows" means "no twin was ever set up", which is a config choice and
    /// not a fault.
    pub configured: bool,
}

impl TwinHealth {
    /// **No standby is answering, and one should be.** The condition a rebuild
    /// keys on. Rows that read `ok` or `behind` are heard;
    /// [`TwinFill::visible`] is the one reader of those words.
    #[must_use]
    pub fn lost(&self) -> bool {
        self.configured && !self.rows.iter().any(TwinFill::visible)
    }
    /// The longest any standby has been unheard, in unix ms, or `None` when
    /// one is answering now or none was ever heard.
    #[must_use]
    pub fn quietest_since(&self) -> Option<i64> {
        if !self.lost() {
            return None;
        }
        self.rows.iter().map(|r| r.last_seen_unix_ms).filter(|t| *t > 0).max()
    }
}

/// What [`ApplianceTwinship::adopt_twin`] did.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct TwinAdopted {
    /// The name the standby is registered under on the primary.
    pub name: String,
    /// The key `BecomeTwin` minted, as enrolled. Public, so it is safe here.
    pub public_key: String,
    /// It is following, as against merely registered.
    pub following: bool,
    /// It came up behind a retention hole and needs `SnapshotFetch`. There is
    /// no shipped driver for that (`bench estate` row B8 is UNRUN by
    /// construction for exactly this), so the flag is how an operator learns
    /// they must run it by hand.
    pub needs_snapshot: bool,
    pub detail: String,
}

/// The phases a product reports, as `DATA-SET-GROWTH-FLOW.md` names them.
/// Kept as words on the wire (`GrowStatus::phase`); this is the reader.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum GrowPhase {
    Idle,
    Draining,
    Unmounted,
    WaitingMembers,
    Growing,
    Mounting,
    Done,
    Failed,
    /// A word this monetize does not know — a newer product. Shown, never
    /// acted on.
    Other,
}

impl GrowPhase {
    pub fn parse(word: &str) -> GrowPhase {
        match word {
            "idle" => GrowPhase::Idle,
            "draining" => GrowPhase::Draining,
            "unmounted" => GrowPhase::Unmounted,
            "waiting-members" => GrowPhase::WaitingMembers,
            "growing" => GrowPhase::Growing,
            "mounting" => GrowPhase::Mounting,
            "done" => GrowPhase::Done,
            "failed" => GrowPhase::Failed,
            _ => GrowPhase::Other,
        }
    }
    /// The set is off the stripe: the outside actor may touch the volumes.
    pub fn volumes_free(self) -> bool {
        matches!(self, GrowPhase::Unmounted | GrowPhase::WaitingMembers)
    }
    pub fn is_terminal(self) -> bool {
        matches!(self, GrowPhase::Done | GrowPhase::Failed | GrowPhase::Idle)
    }
}

/// One member of the set as the product last surveyed it.
#[derive(Clone, PartialEq, Eq, Debug, Default, serde::Serialize, serde::Deserialize)]
pub struct GrowMember {
    pub index: u32,
    pub device: String,
    pub disk_bytes: u64,
    pub set_bytes: u64,
    pub larger: bool,
}

/// The product's answer to [`ApplianceGrow::status`].
#[derive(Clone, PartialEq, Eq, Debug, Default, serde::Serialize, serde::Deserialize)]
pub struct GrowStatus {
    /// `idle` | `draining` | `unmounted` | `waiting-members` | `growing` |
    /// `mounting` | `done` | `failed`. Read with [`GrowPhase::parse`].
    pub phase: String,
    /// `failed`'s reason; empty otherwise.
    pub why: String,
    pub epoch: u64,
    pub target_sectors_per_member: u64,
    pub members: Vec<GrowMember>,
    pub members_verdict: String,
    pub engine_present: bool,
    pub serving: bool,
    pub in_flight: u64,
    pub since_unix_ms: i64,
    pub detail: String,
    pub go_ahead_signer: String,
    pub set_uuid: String,
    pub total_bytes: u64,
}

impl GrowStatus {
    pub fn phase(&self) -> GrowPhase {
        GrowPhase::parse(&self.phase)
    }
}

/// The product's answer to [`ApplianceGrow::flush`]: is the twin caught up,
/// and what does that prove.
#[derive(Clone, PartialEq, Eq, Debug, Default, serde::Serialize, serde::Deserialize)]
pub struct FlushReport {
    pub caught_up: bool,
    pub in_flight: u64,
    pub pending: u64,
    pub bus_gaps: u64,
    pub needs_full_resync: bool,
    pub last_success_unix_ms: i64,
    pub flushed_at_unix_ms: i64,
    pub waited_ms: u64,
    pub catalog_repos: u64,
    pub standbys: u64,
    pub verdict: String,
}

/// What [`Product::purge_tenant`] did.
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct Purged {
    pub tenant: TenantId,
    /// What the product removed, in its own words and its own units — "5 of 5
    /// stores, 1.2 GiB". Free text because every product counts different
    /// things and a shared schema would force each of them to lie a little.
    pub detail: String,
    /// Bytes the product says it reclaimed. `None` means it does not count
    /// them, which is not the same as zero and must not be rendered as it.
    pub bytes_reclaimed: Option<u64>,
    /// **Is the tenant still locked out of the product?**
    ///
    /// `true` is a real and expected outcome — an operator may have suspended
    /// the account separately, and the purge correctly refuses to lift a
    /// lockout it did not take — but it is also what a purge that could not
    /// unlock reports, and either way the tenant cannot use what is left.
    /// Carried so a super purge can say so rather than infer it.
    pub still_locked: bool,
    /// One line per thing the product could not remove. Non-empty means the
    /// tenant's data is PARTIALLY there, and the caller must not go on to
    /// destroy the iron it is sitting on.
    pub failures: Vec<String>,
}

/// **Bytes in the largest unit that still leaves a digit before the decimal
/// point**, by integer arithmetic: `10.0 GiB`, `64.0 MiB`, `999 B`.
///
/// One writer, because there used to be three and every one of them carried the
/// same bug. Each divided by 1 GiB unconditionally and printed one decimal, so
/// anything under a gibibyte rendered `0.0 GiB` — and `0.0` beside a quantity is
/// read as *there is none*:
///
/// * the console (then a Plans page) showed a real 64 MiB
///   `pack_bytes` cap as `0.0 GiB`, which an operator reads as "no quota";
/// * the old catalogue validator said *"sells pack_bytes = 10.0 GiB but
///   its resources buy only 10.0 GiB — short 0.0 GiB"*, a sentence in which
///   every number is wrong in the direction of "nothing is the matter";
/// * `products/gunnar`'s capacity refusals said "short 0.0 GiB" for the same
///   reason.
///
/// It lives here because this is the crate that already owns the meter
/// vocabulary ([`Meter`], [`Backing::DiskBytes`]) and the only one every side
/// can depend on: core, the product plugins, and the browser console alike. It
/// pulls in nothing (this crate is `serde` and nothing else), so the wasm
/// bundle pays a few hundred bytes for a formatter it was carrying anyway.
///
/// Truncates, never rounds up: a reading must not appear to cross a cap it has
/// not crossed.
pub fn bytes(n: u64) -> String {
    const KIB: u64 = 1 << 10;
    const MIB: u64 = 1 << 20;
    const GIB: u64 = 1 << 30;
    const TIB: u64 = 1 << 40;
    let (unit, per) = match n {
        n if n >= TIB => ("TiB", TIB),
        n if n >= GIB => ("GiB", GIB),
        n if n >= MIB => ("MiB", MIB),
        n if n >= KIB => ("KiB", KIB),
        // Under a kibibyte there is nothing to scale to, and `0.0 KiB` would be
        // the same lie one unit down. Bytes are exact and short.
        n => return format!("{n} B"),
    };
    format!("{}.{} {unit}", n / per, ((n % per) * 10) / per)
}

#[cfg(test)]
mod backing_tests {
    use super::Backing;

    /// **A byte over the GiB is sold the next GiB, and a millicore is priced
    /// as a thousandth of a core.** Distinct non-zero numbers: 2200 öre per
    /// GiB-month, 90 GiB + 1 byte, 1500 millicores at 30 000.
    #[test]
    fn prices_round_bytes_up_to_the_unit_and_cpu_by_the_millicore() {
        const GIB: u64 = 1 << 30;
        assert_eq!(Backing::DiskBytes.price_month(90 * GIB, 2200), 198_000, "90 GiB × 22.00 SEK");
        assert_eq!(Backing::DiskBytes.price_month(90 * GIB + 1, 2200), 200_200, "one byte over is the 91st GiB");
        assert_eq!(Backing::DiskBytes.price_month(0, 2200), 0);
        assert_eq!(Backing::RamBytes.price_month(3 * GIB, 700), 2100);
        assert_eq!(Backing::CpuMillicores.price_month(1500, 30_000), 45_000, "1.5 cores at 300.00");
        assert_eq!(Backing::CpuMillicores.price_month(1, 30_000), 30, "one millicore is not free and not a core");
        // A zero unit price is FREE, whatever the delta: the open-source list.
        for b in Backing::ALL {
            assert_eq!(b.price_month(u64::MAX / 4, 0), 0, "{b:?}");
        }
        // The unit names are the price list's keys, both ways.
        for b in Backing::ALL {
            assert_eq!(Backing::from_unit(b.unit()), Some(b));
        }
        assert_eq!(Backing::from_unit("moon_month"), None);
    }
}

#[cfg(test)]
mod bytes_tests {
    use super::bytes;

    /// **A quantity that exists never renders as zero.** RED before this
    /// function existed: `0.0 GiB` for every value under 2^30, in three separate
    /// copies of the same six lines.
    #[test]
    fn only_a_genuine_zero_reads_as_zero() {
        for n in [1u64, 512, 1 << 20, 67_108_864, (1 << 30) - 1] {
            let s = bytes(n);
            assert!(!s.starts_with("0.0 ") && !s.starts_with("0 "), "{n} bytes rendered as {s:?}, which reads as nothing");
        }
        assert_eq!(bytes(0), "0 B");
    }

    #[test]
    fn the_unit_scales_and_the_value_truncates() {
        assert_eq!(bytes(999), "999 B");
        assert_eq!(bytes(1536), "1.5 KiB");
        assert_eq!(bytes(67_108_864), "64.0 MiB");
        assert_eq!(bytes((1 << 30) - 1), "1023.9 MiB", "truncates, never rounds up past the cap");
        assert_eq!(bytes(10 << 30), "10.0 GiB");
        assert_eq!(bytes(3 << 40), "3.0 TiB");
        assert_eq!(bytes(u64::MAX), "16777215.9 TiB", "no overflow at the top of the range");
    }
}

#[cfg(test)]
mod actor_tests {
    use super::*;

    /// A plugin written before the tenant path existed: it implements every
    /// method the trait REQUIRES and knows nothing about tickets.
    struct OldPlugin;

    impl Product for OldPlugin {
        fn id(&self) -> &'static str {
            "old"
        }
        fn meters(&self) -> &[Meter] {
            &[]
        }
        fn list_tenants(&self) -> Result<Vec<TenantId>, ProductError> {
            Ok(Vec::new())
        }
        fn read_usage(&self, _tenant: &TenantId) -> Result<Usage, ProductError> {
            Err(ProductError::Refused("no".into()))
        }
        fn push_entitlement(&self, _fact: &EntitlementFact) -> Result<(), ProductError> {
            Ok(())
        }
        fn can_absorb(&self, _tenant: &TenantId, _caps: &BTreeMap<String, u64>) -> Result<Absorb, String> {
            Ok(Absorb::OnHand)
        }
        fn servable(&self) -> Servable {
            Servable::Unmeasured("a test double measures nothing".to_owned())
        }
    }

    /// ★ **Safe by OMISSION, not by remembering.**
    ///
    /// The out-of-tree plugin nobody will edit when this path ships must not
    /// start honouring signed bytes it has no key for. Both halves of the
    /// default say no: it does not advertise the capability, and it refuses
    /// every ticket rather than falling through to an allow.
    #[test]
    fn a_product_that_never_heard_of_a_ticket_refuses_every_one() {
        let p = OldPlugin;
        assert!(!p.tenant_renewal(), "a console must not be told to offer self-service here");
        let verdict = // The literal and not a constant: this crate cannot depend on
        // `monetize-embed` (that would be a cycle), which is exactly why the
        // trait takes the purpose as a `&str`. The default refuses whatever it
        // is handed, so the word decides nothing here.
        p.may_act_for(&TenantId("alice".into()), "renew", &Default::default(), b"anything at all").expect("the default answers rather than erroring");
        let ActorVerdict::Refused(why) = verdict else { panic!("the default must DENY") };
        assert!(!why.trim().is_empty(), "a refusal says why");
    }

    /// ★ **"NO TWIN" AND "A TWIN THAT STOPPED ANSWERING" ARE DIFFERENT
    /// ANSWERS, AND ONLY ONE OF THEM IS A FAULT.**
    ///
    /// An estate that never had a standby is a choice somebody made in the
    /// config; building one unasked would be monetize inventing infrastructure.
    /// An estate that HAS one configured and cannot hear it has lost
    /// something, and that is the only case a rebuild may key on. This is the
    /// same distinction `grow::twin_sizing` makes for a missing twin lane and
    /// `replication_flush` makes with FailedPrecondition.
    #[test]
    fn a_twin_never_configured_is_not_a_lost_twin() {
        let silent = TwinFill { principal: "twin".into(), state: "silent".into(), last_seen_unix_ms: 1_900_000_000_000, ..TwinFill::default() };

        // Configured and silent: lost.
        let h = TwinHealth { rows: vec![silent.clone()], configured: true };
        assert!(h.lost());
        assert_eq!(h.quietest_since(), Some(1_900_000_000_000));

        // The same silence, but nobody ever asked for a twin: not a fault.
        let h = TwinHealth { rows: vec![silent.clone()], configured: false };
        assert!(!h.lost(), "an estate with no standby configured has lost nothing");
        assert_eq!(h.quietest_since(), None);

        // Configured, and no rows at all — the twin is not merely quiet, it is
        // absent from the primary's own health. Still lost.
        let h = TwinHealth { rows: vec![], configured: true };
        assert!(h.lost());
        assert_eq!(h.quietest_since(), None, "never heard is not a time");
    }

    /// **`behind` is heard.** A twin that is catching up is answering, and
    /// replacing it would throw away the catching-up it has already done. Only
    /// `silent`, or a word this build does not know, counts as unheard.
    #[test]
    fn a_twin_that_is_merely_behind_is_still_heard_and_is_never_rebuilt() {
        for word in ["ok", "behind"] {
            let h = TwinHealth { rows: vec![TwinFill { state: word.into(), last_seen_unix_ms: 5, ..TwinFill::default() }], configured: true };
            assert!(!h.lost(), "{word} is heard");
        }
        for word in ["silent", "", "who-knows"] {
            let h = TwinHealth { rows: vec![TwinFill { state: word.into(), last_seen_unix_ms: 5, ..TwinFill::default() }], configured: true };
            assert!(h.lost(), "{word:?} is not heard");
        }
    }

    /// **One standby answering is enough.** A deployment may have two; the
    /// primary has not lost its standby until none of them is heard.
    #[test]
    fn one_standby_answering_is_enough_to_not_be_lost() {
        let h = TwinHealth {
            rows: vec![
                TwinFill { principal: "a".into(), state: "silent".into(), ..TwinFill::default() },
                TwinFill { principal: "b".into(), state: "ok".into(), ..TwinFill::default() },
            ],
            configured: true,
        };
        assert!(!h.lost());
    }
}