crates/monetize-server/src/services/ledger.rs
use std::sync::Arc;
use monetize_proto::ledger_server::Ledger;
use monetize_proto::query_ledger_request::By;
use monetize_proto::{self as pb};
use tonic::{Request, Response, Status};
use super::{blocking, ledger_err, product_or_not_found};
use crate::convert::{money_to_pb, row_to_pb};
use crate::state::AppState;
pub struct LedgerService(pub Arc<AppState>);
#[tonic::async_trait]
impl Ledger for LedgerService {
async fn query(
&self,
req: Request<pb::QueryLedgerRequest>,
) -> Result<Response<pb::QueryLedgerResponse>, Status> {
let by = req.into_inner().by.ok_or_else(|| {
Status::invalid_argument("query needs reference, tenant or unfinished")
})?;
blocking(&self.0, move |s| {
let mut rows = Vec::new();
let mut references = Vec::new();
match by {
By::Reference(r) => {
rows.extend(s.ledger.rows(&r).map_err(ledger_err)?.iter().map(row_to_pb));
}
By::Tenant(t) => {
// References are `<product>/<tenant>/<date>`, and the ledger now has a
// real prefix scan, so a tenant's history is ALL of it.
//
// It used to be reconstructed from the open set plus whatever reference
// the entitlement's `source` happened to name, because the ledger had no
// prefix scan. That could not see a CLOSED reference from an earlier
// period — so this page showed a paying tenant's current year and
// silently dropped every year before it. The teardown lane needed the
// same scan for a harder reason (a give-back that misses a period leaves
// a disk billing forever), and both callers now share one primitive.
// **A string prefix is not a tenant.** `references_with_prefix`
// matches `gunnar/team/sub/…` for a query about `team`, and
// gunnar treats nested namespaces as SEPARATE tenants — so
// without this filter a tenant's ledger, and the billing
// reconciliation built on it, silently included another
// tenant's money. See `is_exactly` for why the test is on
// the reference's shape rather than on a fold.
let refs: Vec<String> = s
.ledger
.references_with_prefix(&format!("{}/{}/", t.product, t.tenant))
.map_err(ledger_err)?
.into_iter()
.filter(|r| is_exactly(r, &t.product, &t.tenant))
.collect();
for r in &refs {
rows.extend(s.ledger.rows(r).map_err(ledger_err)?.iter().map(row_to_pb));
}
references = refs;
}
By::Unfinished(_) => {
references = s.ledger.unfinished().map_err(ledger_err)?;
for r in &references {
rows.extend(s.ledger.rows(r).map_err(ledger_err)?.iter().map(row_to_pb));
}
}
// The one arm whose ORDER is the answer: newest first, across
// every reference, straight off the ledger's time index. See the
// proto for what the console was doing instead, and why a fact is
// not a row.
By::Recent(limit) => {
for (reference, row) in s.ledger.recent(limit as usize).map_err(ledger_err)? {
if !references.contains(&reference) {
references.push(reference);
}
rows.push(row_to_pb(&row));
}
}
}
Ok(Response::new(pb::QueryLedgerResponse { rows, references }))
})
.await
}
/// Write the operational state to the SERVER'S configured backup directory.
/// See the proto for why this is an RPC and not a CLI verb (redb is
/// single-process, so the running service holds the ledger's lock) and why
/// the request carries no path.
async fn export(
&self,
_req: Request<pb::ExportBackupRequest>,
) -> Result<Response<pb::ExportBackupResponse>, Status> {
blocking(&self.0, move |s| {
let dir = s.backup_dir.clone().unwrap_or_default();
let m = crate::backup::scheduled_export(s).map_err(|e| match e {
crate::backup::BackupError::NoBackupDir => {
Status::failed_precondition(e.to_string())
}
e => Status::internal(e.to_string()),
})?;
skidbladnir::functional_status(
"monetize-server/backup",
"export",
true,
// The signup count is on the STATUS LINE and not on
// `ExportBackupResponse`: that message is vendored by the
// console lanes, and a backup's row counts are an operator's
// reassurance rather than something a caller branches on. The
// manifest carries the authoritative number either way.
&format!(
"{} ledger row(s), {} usage reading(s), {} signup(s) -> {}",
m.ledger_rows,
m.usage_readings,
m.signups,
dir.display()
),
);
Ok(Response::new(pb::ExportBackupResponse {
dir: dir.display().to_string(),
ledger_rows: m.ledger_rows as u64,
usage_readings: m.usage_readings as u64,
signing_public_key: m.signing_public_key,
}))
})
.await
}
/// **Recent attempts, folded into outcome lines.** `monetize::Ledger::recent_sales`
/// over the same time index `Query(recent)` uses — see the proto for why this
/// is not that RPC with a different shape.
///
/// The product filter is applied AFTER the fold and the limit is applied
/// after the filter, so asking for gunnar's last ten sales on a server that
/// also serves holger does not return three. The walk is bounded by
/// `Ledger::RECENT_CEILING` rows either way.
async fn sales(
&self,
req: Request<pb::SalesRequest>,
) -> Result<Response<pb::SalesResponse>, Status> {
let r = req.into_inner();
blocking(&self.0, move |s| {
if !r.product.is_empty() {
product_or_not_found(s, &r.product)?;
}
let want = if r.limit == 0 {
usize::MAX
} else {
r.limit as usize
};
let mut sales: Vec<pb::Sale> = s
.ledger
.recent_sales(0, 0)
.map_err(ledger_err)?
.iter()
.filter(|sale| r.product.is_empty() || sale.product == r.product)
.map(sale_to_pb)
.collect();
// **A paid signup is a purchase too**, though it writes no ledger
// row: one line per vendor session, folded from the signup store
// (`crate::signup::signup_sales`), newest first with the rest.
sales.extend(
crate::signup::signup_sales(s, &r.product)
.map_err(|e| Status::internal(format!("signup store: {e}")))?,
);
sales.sort_by_key(|sale| std::cmp::Reverse(sale.at_unix_ms));
sales.truncate(want);
Ok(Response::new(pb::SalesResponse { sales }))
})
.await
}
}
/// One folded attempt on the wire. `outcome` and `reason` are separate fields
/// rather than one string, so a console renders the badge from the first and the
/// detail from the second and neither has to be parsed out of the other.
pub(crate) fn sale_to_pb(s: &monetize::ledger::Sale) -> pb::Sale {
use monetize::ledger::SaleOutcome;
let reason = match &s.outcome {
SaleOutcome::Refunded { reason, .. } | SaleOutcome::Declined { reason } => reason.clone(),
SaleOutcome::Granted | SaleOutcome::Open => String::new(),
};
pb::Sale {
reference: s.reference.clone(),
product: s.product.clone(),
tenant: s.tenant.clone(),
price: Some(money_to_pb(s.price)),
cost: Some(money_to_pb(s.cost)),
// `None` travels as absent, and absent is "nobody said", not "zero".
fee: s.fee.map(money_to_pb),
// A declined attempt charged nothing, so it carries no tax.
tax: (!matches!(s.outcome, SaleOutcome::Declined { .. })).then(|| money_to_pb(s.tax())),
tax_from_vendor: s.tax_from_vendor(),
vat_basis_points: if matches!(s.outcome, SaleOutcome::Declined { .. }) {
0
} else {
s.basis_points()
},
caps: s.caps.clone(),
delta: s.delta.clone(),
started_unix_ms: s.started_unix_ms,
at_unix_ms: s.at_unix_ms,
steps: s.steps as u32,
drawn: s.drawn,
outcome: s.outcome.name().into(),
reason,
needs_you: s.outcome.needs_you(),
livemode: s.livemode,
// The buyer's country as the vendor returned it for this payment.
customer_country: s.customer_country.clone().unwrap_or_default(),
}
}
/// **Is this reference EXACTLY this tenant's, rather than a nested tenant's?**
///
/// `references_with_prefix("gunnar/team/")` is a *string* prefix, and a gunnar
/// namespace may itself contain a slash. `gunnar/team/sub/2026-09-17+…` starts
/// with that prefix and belongs to a different tenant — gunnar says so in
/// writing: *"a grant on `team` does not reach `team/sub`. Nested namespaces
/// are separate tenants, and inheritance would be a privilege escalation that
/// nobody wrote down."* So the scan returned `team/sub`'s rows for a query
/// about `team`, and a tenant's billing reconciliation silently included
/// another tenant's money.
///
/// # Why this is structural and not a fold
///
/// `Purchase.History` answers the same question by folding each attempt and
/// comparing `Sale::tenant` — the authoritative name off the `Reserved` row.
/// That is the better answer THERE, because that call returns folded sales and
/// an attempt that does not fold has nothing to return anyway.
///
/// It is the wrong answer HERE. `Sale::fold` is **not total**: an attempt
/// refused at reserve has rows and no `Reserved` step, so it folds to `None`.
/// This call returns RAW rows, so filtering by the fold would silently stop
/// showing an operator every attempt that was refused before it opened —
/// fixing a leak by hiding evidence.
///
/// So the test is on the reference's own shape, which
/// `monetize::transaction::reference_of` writes as
/// `<product>/<tenant>/<YYYY-MM-DD>+<caps>`: after the prefix, a reference that
/// really is this tenant's has **no further `/`**. The date has none, and
/// `caps_fingerprint` is `meter=value` joined by `,`. A nested tenant's
/// remainder begins `sub/`, and that is the whole difference.
pub(crate) fn is_exactly(reference: &str, product: &str, tenant: &str) -> bool {
let prefix = format!("{product}/{tenant}/");
match reference.strip_prefix(&prefix) {
Some(rest) => !rest.is_empty() && !rest.contains('/'),
None => false,
}
}
#[cfg(test)]
mod tests {
use super::is_exactly;
/// ★ **A query about `team` must not answer with `team/sub`'s rows.**
#[test]
fn a_nested_namespace_is_a_different_tenant() {
assert!(is_exactly(
"gunnar/team/2026-09-17+pack_bytes=10",
"gunnar",
"team"
));
assert!(
!is_exactly("gunnar/team/sub/2026-09-17+pack_bytes=10", "gunnar", "team"),
"a nested namespace's reference was counted as the parent's"
);
// …and the nested tenant's own query still finds its own.
assert!(is_exactly(
"gunnar/team/sub/2026-09-17+pack_bytes=10",
"gunnar",
"team/sub"
));
}
/// The caps fingerprint is `meter=value` joined by `,`, and the date is
/// ISO — neither contains a slash, so a real reference's remainder never
/// does. Several meters, which is the case most likely to grow a separator
/// somebody did not think about.
#[test]
fn a_many_meter_reference_is_still_exactly_its_tenants() {
assert!(is_exactly(
"gunnar/alice/2026-09-17+cache_bytes=1,pack_bytes=10,repos=3",
"gunnar",
"alice"
));
}
/// Another product, and a tenant whose name merely starts the same way.
#[test]
fn a_prefix_that_is_not_a_path_boundary_is_not_a_match() {
assert!(!is_exactly(
"holger/alice/2026-09-17+x=1",
"gunnar",
"alice"
));
assert!(
!is_exactly("gunnar/alicia/2026-09-17+x=1", "gunnar", "alice"),
"`alicia` is not `alice`, and the trailing slash in the prefix is what says so"
);
}
/// A bare prefix with nothing after it names no attempt.
#[test]
fn a_reference_that_is_only_the_prefix_is_not_an_attempt() {
assert!(!is_exactly("gunnar/alice/", "gunnar", "alice"));
}
}