crates/monetize-ui-server/src/grpc.rs
//! The client half: one lazy channel to monetize-server, the suite bearer on
//! every call. Ten generated clients over one `InterceptedService`. Nothing is
//! cached: monetize's reads are local redb lookups,
//! and since `Events.Watch` landed the console does not re-read on a timer at
//! all while the stream is up.
use monetize_proto::accounts_client::AccountsClient;
use monetize_proto::callback_admin_client::CallbackAdminClient;
use monetize_proto::entitlement_client::EntitlementClient;
use monetize_proto::estate_admin_client::EstateAdminClient;
use monetize_proto::estate_order_client::EstateOrderClient;
use monetize_proto::estates_client::EstatesClient;
use monetize_proto::events_client::EventsClient;
use monetize_proto::guardian_client::GuardianClient;
use monetize_proto::ledger_client::LedgerClient;
use monetize_proto::plugins_client::PluginsClient;
use monetize_proto::products_client::ProductsClient;
use monetize_proto::purchase_client::PurchaseClient;
use monetize_proto::reconcile_admin_client::ReconcileAdminClient;
use monetize_proto::signup_admin_client::SignupAdminClient;
use monetize_proto::site_book_client::SiteBookClient;
use monetize_proto::tax_client::TaxClient;
use monetize_proto::test_purchase_client::TestPurchaseClient;
use monetize_proto::usage_client::UsageClient;
use tonic::metadata::{Ascii, MetadataValue};
use tonic::service::interceptor::InterceptedService;
use tonic::transport::Channel;
/// **How this console proves who it is to one monetize, on every call.**
///
/// * `Bearer` — `authorization: Bearer <token>`: the console INSIDE the
/// estate, on loopback (nothing when the bearer is empty: an open dev
/// server);
/// * `Sshsig` — the shared operator console OUTSIDE the estates (owner
/// 2026-09-28): each call signed with the console's own SSH key for THAT
/// method (`monetize-authorization`, `monetize_proto::control_auth`), fresh
/// nonce, never a bearer. The estate's monetize knows only the public half
/// (the `monetize-ui` principal).
#[derive(Clone)]
pub enum CallAuth {
Bearer(Option<MetadataValue<Ascii>>),
Sshsig(std::sync::Arc<monetize_proto::control_auth::sshsig::ssh_key::PrivateKey>),
}
/// Kept as the name the rest of this crate and its tests use.
pub type Bearer = CallAuth;
impl tonic::service::Interceptor for CallAuth {
fn call(&mut self, mut req: tonic::Request<()>) -> Result<tonic::Request<()>, tonic::Status> {
match self {
CallAuth::Bearer(Some(v)) => {
req.metadata_mut().insert("authorization", v.clone());
}
CallAuth::Bearer(None) => {}
CallAuth::Sshsig(key) => {
// The generated client names the method it is calling; the
// credential is bound to exactly that.
let method = req
.extensions()
.get::<tonic::GrpcMethod>()
.map(|m| {
format!(
"{}.{}",
m.service().rsplit('.').next().unwrap_or(m.service()),
m.method()
)
})
.ok_or_else(|| {
tonic::Status::internal(
"no gRPC method on the request; nothing to sign for",
)
})?;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0);
let nonce = uuid::Uuid::new_v4().simple().to_string();
let c =
monetize_proto::control_auth::sign(key, now, &nonce, &method).map_err(|e| {
tonic::Status::internal(format!("signing the call failed: {e}"))
})?;
let v = MetadataValue::try_from(c.to_header_value()).map_err(|e| {
tonic::Status::internal(format!("the credential is not a header value: {e}"))
})?;
req.metadata_mut()
.insert(monetize_proto::control_auth::METADATA_KEY, v);
}
}
Ok(req)
}
}
pub type Svc = InterceptedService<Bounded, CallAuth>;
/// **The channel with every call bounded whole** (A22 #6, F-F-09), its connection included:
/// a lazy channel connects (TCP and TLS) while a call waits, and neither the channel's own
/// `timeout` (it starts once the connection is ready) nor its `connect_timeout` (TCP only)
/// bounds a TLS handshake that never finishes. Past [`CALL_WITHIN`] the call is
/// `DEADLINE_EXCEEDED` (the console's 504), and nothing of it is waited on any more.
#[derive(Clone)]
pub struct Bounded {
channel: Channel,
within: std::time::Duration,
}
impl tower::Service<http::Request<tonic::body::Body>> for Bounded {
type Response = http::Response<tonic::body::Body>;
type Error = Box<dyn std::error::Error + Send + Sync>;
type Future = std::pin::Pin<
Box<dyn std::future::Future<Output = Result<Self::Response, Self::Error>> + Send>,
>;
fn poll_ready(
&mut self,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Result<(), Self::Error>> {
self.channel.poll_ready(cx).map_err(Into::into)
}
fn call(&mut self, req: http::Request<tonic::body::Body>) -> Self::Future {
let within = self.within;
let call = self.channel.call(req);
Box::pin(async move {
match tokio::time::timeout(within, call).await {
Ok(answered) => answered.map_err(Into::into),
Err(_) => Err(Box::new(tonic::Status::deadline_exceeded(format!(
"monetize did not answer within {} s",
within.as_secs_f64()
))) as Self::Error),
}
})
}
}
fn bearer(token: &str) -> Result<CallAuth, String> {
Ok(CallAuth::Bearer(match token {
"" => None,
t => Some(
MetadataValue::try_from(format!("Bearer {t}"))
.map_err(|e| format!("the bearer is not a header value: {e}"))?,
),
}))
}
/// The console's SSH key (OpenSSH private key, unencrypted) — the identity
/// every estate knows as `monetize-ui`.
pub fn load_key(path: &std::path::Path) -> Result<CallAuth, String> {
let text = std::fs::read_to_string(path).map_err(|e| format!("{}: {e}", path.display()))?;
let key = monetize_proto::control_auth::sshsig::ssh_key::PrivateKey::from_openssh(text.trim())
.map_err(|e| format!("{}: not an OpenSSH private key: {e}", path.display()))?;
Ok(CallAuth::Sshsig(std::sync::Arc::new(key)))
}
/// How long the console waits for a monetize's TCP connection (A22 #6, F-F-09): a box that
/// is down is a status within seconds, never a page that waits. A TLS handshake that never
/// finishes is bounded by [`CALL_WITHIN`] (it is part of the call).
pub const CONNECT_WITHIN: std::time::Duration = std::time::Duration::from_secs(10);
/// The most one call may take, its connection included (A22 #6, F-F-09). A bound, not a
/// budget: monetize answers its long work at once (a terminate, a plugin update, a guardian
/// press is only approved) and the console polls; the slowest call left is an update's
/// start, one firewall write at the provider. A stream (`Events.Watch`) is not cut: the
/// bound is on its first answer.
pub const CALL_WITHIN: std::time::Duration = std::time::Duration::from_secs(120);
/// [`CONNECT_WITHIN`] and [`CALL_WITHIN`], as one channel is given them.
#[derive(Clone, Copy, Debug)]
struct Patience {
connect: std::time::Duration,
call: std::time::Duration,
}
const CONSOLE: Patience = Patience {
connect: CONNECT_WITHIN,
call: CALL_WITHIN,
};
#[derive(Clone)]
pub struct Monetize {
channel: Bounded,
bearer: CallAuth,
/// **The last fleet this console read whole** (PROD-4): when, and what — answered with its
/// age when a later read does not come within [`crate::estates::FLEET_WITHIN`].
pub(crate) fleet_seen: crate::bounded::Seen<monetize_proto::GuardianCasesResponse>,
/// The last map of each estate drawn, and each product's machines at the cloud (PROD-4).
pub(crate) draws_seen: crate::bounded::Seen<monetize_proto::DrawEstateResponse>,
pub(crate) clouds_seen: crate::bounded::Seen<monetize_proto::FleetResponse>,
/// Each operator site's last status, for its first-admin door on the Sites page (MHUNT-1 /
/// RA5-25): asked once per site at a time, within the page's bound, and kept a while.
pub(crate) site_status_seen: crate::bounded::Seen<monetize_proto::EstateOrderStatusResponse>,
/// The last Overview the box classified (OVW-CLEAN), answered with its age past the bound.
pub(crate) estates_seen: crate::bounded::Seen<monetize_proto::EstatesOverviewResponse>,
/// The jera job that keeps [`Self::fleet_seen`] (RA3-1).
pub(crate) fleet_keeper: crate::estates::FleetKeeper,
}
impl Monetize {
/// **Start the jera job that keeps the fleet** (RA3-1) at the console's start, so the
/// first page after a restart finds a fleet read already out or kept. `Err` with no runtime.
pub fn keep_the_fleet(&self) -> Result<(), String> {
self.fleet_keeper.lane(self).map(|_| ())
}
/// `addr` is `host:port` or a full `http://` URI. Connects on first use.
pub fn connect_lazy(addr: &str, token: &str) -> Result<Monetize, String> {
Self::connect_lazy_within(addr, token, CONSOLE)
}
fn connect_lazy_within(
addr: &str,
token: &str,
patience: Patience,
) -> Result<Monetize, String> {
let uri = if addr.contains("://") {
addr.to_owned()
} else {
format!("http://{addr}")
};
let channel = Channel::from_shared(uri)
.map_err(|e| format!("bad control address {addr:?}: {e}"))?
.connect_timeout(patience.connect)
.connect_lazy();
Ok(Monetize {
channel: Bounded {
channel,
within: patience.call,
},
bearer: bearer(token)?,
fleet_seen: Default::default(),
draws_seen: Default::default(),
clouds_seen: Default::default(),
site_status_seen: Default::default(),
estates_seen: Default::default(),
fleet_keeper: Default::default(),
})
}
/// **monetize reached over TLS, pinned** — an estate's `monetize.<apex>`
/// (its front proxies the control API to monetize on loopback). `url` is
/// `https://<name>[:port]`: the name is the TLS server name AND the
/// request authority the front routes by. `at` dials another address
/// (`host:port`) under that name — the mock's loopback forward of vm1's
/// :443. `ca_pem` is the ONLY trust: the estate's own certificate (or its
/// issuer), never the public roots, so a console that was told which
/// estate it administers cannot be handed another.
pub fn connect_pinned(
url: &str,
at: Option<&str>,
ca_pem: &str,
token: &str,
) -> Result<Monetize, String> {
Self::connect_pinned_as(url, at, ca_pem, bearer(token)?)
}
/// [`Monetize::connect_pinned`] proving itself with `auth` (the console's
/// SSH key, [`load_key`], outside the estate).
pub fn connect_pinned_as(
url: &str,
at: Option<&str>,
ca_pem: &str,
auth: CallAuth,
) -> Result<Monetize, String> {
Self::connect_pinned_within(url, at, ca_pem, auth, CONSOLE)
}
fn connect_pinned_within(
url: &str,
at: Option<&str>,
ca_pem: &str,
auth: CallAuth,
patience: Patience,
) -> Result<Monetize, String> {
use tonic::transport::{Certificate, ClientTlsConfig, Endpoint};
let origin: http::Uri = url
.parse()
.map_err(|e| format!("bad control url {url:?}: {e}"))?;
if origin.scheme_str() != Some("https") {
return Err(format!(
"a pinned control url is https://<name>, not {url:?}"
));
}
let name = origin
.host()
.ok_or_else(|| format!("{url:?} names no host"))?
.to_owned();
let port = origin.port_u16().unwrap_or(443);
let dial = match at {
Some(a) => format!("https://{a}"),
None => format!("https://{name}:{port}"),
};
let tls = ClientTlsConfig::new()
.ca_certificate(Certificate::from_pem(ca_pem))
.domain_name(name.clone());
let channel = Endpoint::from_shared(dial.clone())
.map_err(|e| format!("bad control address {dial:?}: {e}"))?
.origin(origin)
.tls_config(tls)
.map_err(|e| format!("the pinned TLS for {name} does not build: {e}"))?
.connect_timeout(patience.connect)
.connect_lazy();
Ok(Monetize {
channel: Bounded {
channel,
within: patience.call,
},
bearer: auth,
fleet_seen: Default::default(),
draws_seen: Default::default(),
clouds_seen: Default::default(),
site_status_seen: Default::default(),
estates_seen: Default::default(),
fleet_keeper: Default::default(),
})
}
/// **Which key this console signs this estate's calls with**, as its
/// public SHA-256 fingerprint (`SHA256:…`) — what an operator compares with
/// the `monetize-ui` principal the estate was fed. `None` on a bearer or an
/// open loopback control. Never the key.
pub fn signing_fingerprint(&self) -> Option<String> {
match &self.bearer {
CallAuth::Sshsig(k) => Some(
k.public_key()
.fingerprint(monetize_proto::control_auth::sshsig::ssh_key::HashAlg::Sha256)
.to_string(),
),
CallAuth::Bearer(_) => None,
}
}
fn svc(&self) -> Svc {
InterceptedService::new(self.channel.clone(), self.bearer.clone())
}
pub fn entitlement(&self) -> EntitlementClient<Svc> {
EntitlementClient::new(self.svc())
}
/// The private estates an estate SOLD (`EstateOrder.List`): the operator's
/// read, each order naming its account (one shop, 2026-10-07).
pub fn estate_orders(&self) -> EstateOrderClient<Svc> {
// The console call's body is an artifact (up to 32 MiB), its answer too (lane OX-WGDS4).
EstateOrderClient::new(self.svc())
.max_encoding_message_size(ESTATE_ORDER_MESSAGE_MAX)
.max_decoding_message_size(ESTATE_ORDER_MESSAGE_MAX)
}
/// **Every (product, tenant) account and its rows** (`Accounts`, ONE
/// monetize 5.4). Read-only, same bearer as `Ledger`.
pub fn accounts(&self) -> AccountsClient<Svc> {
AccountsClient::new(self.svc())
}
/// **The estate guardian's console** (`Guardian`, 5.4): cases, the four
/// buttons and the auto-fix switch. A press carries the principal this BFF
/// vouches for; monetize never reaches into an estate on it.
pub fn guardian(&self) -> GuardianClient<Svc> {
GuardianClient::new(self.svc())
}
pub fn usage(&self) -> UsageClient<Svc> {
UsageClient::new(self.svc())
}
pub fn products(&self) -> ProductsClient<Svc> {
ProductsClient::new(self.svc())
}
pub fn ledger(&self) -> LedgerClient<Svc> {
LedgerClient::new(self.svc())
}
pub fn purchase(&self) -> PurchaseClient<Svc> {
PurchaseClient::new(self.svc())
}
/// The operator's read of the PAID SIGNUP store — every checkout session a
/// vendor's webhook brought in, refusals included. The console never speaks
/// `Signup` (the customer-facing poll): that one is the product's own return
/// page's, and this BFF serves operators.
pub fn signups(&self) -> SignupAdminClient<Svc> {
SignupAdminClient::new(self.svc())
}
/// **The REFUSED VENDOR CALLBACKS, read side.** A bounded ring in monetize,
/// holding the deliveries it threw away — the failure that used to leave no
/// trace but a journald line. Same bearer as the rest: the refusal is a
/// caller who did not have one.
pub fn callbacks(&self) -> CallbackAdminClient<Svc> {
CallbackAdminClient::new(self.svc())
}
/// **What the VENDOR says it took and monetize did not fulfil, read
/// side.** The reconciler's durable rows — the payments behind the refused
/// callbacks, read back from the vendor over monetize's own key. Same
/// bearer as the rest.
pub fn reconcile(&self) -> ReconcileAdminClient<Svc> {
ReconcileAdminClient::new(self.svc())
}
/// **The TAX RETURN, read side.** The periods a year is filed in, and one
/// filled form. Behind the same bearer as `Ledger`, because it is the same
/// ledger read through a jurisdiction's boxes.
pub fn tax(&self) -> TaxClient<Svc> {
TaxClient::new(self.svc())
}
/// **The ESTATE, drawn.** One board, one chip per machine, rendered in
/// monetize because that is the only process that holds both halves of it:
/// the `[machines]` declaration and whatever the product plugin read off
/// the live boxes. This console does not re-draw it and must not — a second
/// renderer on this side of the wire would be free to disagree about which
/// number was measured and which was merely claimed, which is the one
/// mistake the picture exists to prevent. Same bearer as the rest.
pub fn estate(&self) -> EstateAdminClient<Svc> {
EstateAdminClient::new(self.svc())
}
/// The live feed. One long-lived stream per process, held by
/// [`crate::sse::EventHub`] — not per browser: a hundred open tabs must not
/// become a hundred subscriptions on monetize's ring.
pub fn events(&self) -> EventsClient<Svc> {
EventsClient::new(self.svc())
}
/// **The test button** (`TestPurchase`, 2026-10-11): the operator's door
/// only — monetize refuses the service bearer on it.
pub fn test_purchase(&self) -> TestPurchaseClient<Svc> {
TestPurchaseClient::new(self.svc())
}
/// **The operator's site book** (`SiteBook`, 2026-10-14): a site added and edited in the
/// application — the operator's door only, as `EstateOrder.OrderSite`.
pub fn site_book(&self) -> SiteBookClient<Svc> {
SiteBookClient::new(self.svc())
}
/// **The box's plugins** (`Plugins`, 2026-10-17): the release's signed plugins taken from
/// the release bucket as a queued job — the operator's door only, as `SiteBook`.
pub fn plugins(&self) -> PluginsClient<Svc> {
PluginsClient::new(self.svc())
}
/// **The estates that exist, and the New-estate slot** (`Estates`, OVW-CLEAN): the Overview
/// behind the console's bearer; `NewEstate` the operator's door only, as `OrderSite`.
pub fn estates(&self) -> EstatesClient<Svc> {
EstatesClient::new(self.svc())
}
}
/// A refusal, in monetize's own words. `code` is the gRPC status lowercased
/// (`notfound`, `invalidargument`, …) so the console can show it verbatim.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub struct Fail {
pub code: String,
pub message: String,
}
impl From<tonic::Status> for Fail {
fn from(s: tonic::Status) -> Fail {
Fail {
code: format!("{:?}", s.code()).to_ascii_lowercase(),
message: s.message().to_owned(),
}
}
}
impl Fail {
pub fn invalid(message: impl Into<String>) -> Fail {
Fail {
code: "invalidargument".into(),
message: message.into(),
}
}
/// The HTTP status a gRPC refusal becomes. Upstream refusals of the
/// console's OWN bearer are 502, not 401: the browser's session is fine.
pub fn http_status(&self) -> http::StatusCode {
use http::StatusCode as S;
match self.code.as_str() {
"invalidargument" | "failedprecondition" | "outofrange" | "confirm" => S::BAD_REQUEST,
"notfound" => S::NOT_FOUND,
"alreadyexists" | "aborted" => S::CONFLICT,
"permissiondenied" => S::FORBIDDEN,
"unimplemented" => S::NOT_IMPLEMENTED,
"unavailable" => S::SERVICE_UNAVAILABLE,
"deadlineexceeded" => S::GATEWAY_TIMEOUT,
"resourceexhausted" => S::PAYLOAD_TOO_LARGE,
_ => S::BAD_GATEWAY,
}
}
}
/// **The `EstateOrder` messages' limit** on this client: monetize-server's
/// `estate_order::product_order::GRPC_MESSAGE_MAX` (a product order's 32 MiB file and its fields).
pub const ESTATE_ORDER_MESSAGE_MAX: usize = (32 << 20) + (1 << 20);
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn a_status_keeps_its_words_and_maps_to_the_right_http_status() {
let f: Fail = tonic::Status::not_found("product \"x\"").into();
assert_eq!(
f,
Fail {
code: "notfound".into(),
message: "product \"x\"".into()
}
);
assert_eq!(f.http_status(), http::StatusCode::NOT_FOUND);
assert_eq!(
Fail::from(tonic::Status::unauthenticated("bearer")).http_status(),
http::StatusCode::BAD_GATEWAY
);
assert!(Monetize::connect_lazy("127.0.0.1:50061", "").is_ok());
assert!(Monetize::connect_lazy("127.0.0.1:50061", "tok\nen").is_err());
}
/// A listener that takes every connection and never says a word (a monetize that hangs).
fn silent() -> std::net::SocketAddr {
let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let at = l.local_addr().unwrap();
std::thread::spawn(move || {
let mut held = Vec::new();
for s in l.incoming() {
held.push(s);
}
});
at
}
/// A short bound for the tests (the console's are [`CONNECT_WITHIN`] and [`CALL_WITHIN`]).
const SHORT: Patience = Patience {
connect: CONNECT_WITHIN,
call: std::time::Duration::from_millis(300),
};
/// ★ A22 #6 (F-F-09): **a pinned estate that takes the connection and never answers its
/// TLS gives the console a status within the call bound**, never a page that hangs.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_pinned_monetize_that_never_answers_its_handshake_gives_a_status_in_time() {
let at = silent();
let ca = rcgen::generate_simple_self_signed(vec!["monetize.example".to_string()])
.unwrap()
.cert
.pem();
let m = Monetize::connect_pinned_within(
"https://monetize.example",
Some(&at.to_string()),
&ca,
bearer("").unwrap(),
SHORT,
)
.unwrap();
let started = std::time::Instant::now();
let asked = tokio::time::timeout(
std::time::Duration::from_secs(30),
m.products().list(monetize_proto::ListProductsRequest {}),
)
.await;
assert!(
matches!(&asked, Ok(Err(st)) if st.code() == tonic::Code::DeadlineExceeded),
"DEADLINE_EXCEEDED, not a hang: {asked:?} after {:?}",
started.elapsed()
);
assert!(
started.elapsed() < std::time::Duration::from_secs(5),
"within the bound: {:?}",
started.elapsed()
);
}
/// The console's own bounds are the ones named, on every constructor.
#[test]
fn the_console_waits_ten_seconds_for_a_connection_and_two_minutes_for_a_call() {
assert_eq!(
(CONSOLE.connect, CONSOLE.call),
(
std::time::Duration::from_secs(10),
std::time::Duration::from_secs(120)
)
);
}
/// ★ A22 #6 (F-F-09): **a monetize that takes the call and never answers it gives the
/// console a status within the call bound** (here a short one; the console's is
/// [`CALL_WITHIN`]).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_monetize_that_never_answers_a_call_gives_a_status_within_the_call_bound() {
let at = silent();
let m = Monetize::connect_lazy_within(&at.to_string(), "", SHORT).unwrap();
let started = std::time::Instant::now();
let asked = tokio::time::timeout(
std::time::Duration::from_secs(10),
m.products().list(monetize_proto::ListProductsRequest {}),
)
.await;
assert!(
matches!(&asked, Ok(Err(st)) if st.code() == tonic::Code::DeadlineExceeded),
"DEADLINE_EXCEEDED, not a hang: {asked:?} after {:?}",
started.elapsed()
);
assert!(
started.elapsed() < std::time::Duration::from_secs(5),
"within the bound: {:?}",
started.elapsed()
);
}
}