flagon-io/g1t

public

1	//! The billing service: what g1t's compute costs, charged to the workspace
2	//! it ran for at what g1t pays plus 20%.
3	//!
4	//! One paid plan, "g1t" (see `features`): $20 a month per workspace, never
5	//! per person, with $10 of usage included, deployments, and more private
6	//! storage. The forge is free for everyone; compute needs the plan or a
7	//! card check (see `compute` and `cards`). Before anything that costs money
8	//! starts, the service that starts it reserves its estimate here; when it
9	//! is done, what it cost goes on the ledger plus the margin, drawn first
10	//! from what the plan includes, a trial or g1t's pools (see `credits`).
11	//! Every change is a ledger entry, and a balance is always the sum of its
12	//! ledger.
13	//!
14	//! Without a card processor configured the service says so and charges
15	//! nothing, so that g1t still runs where billing has not been set up.
16	//!
17	//! Reached only through service bindings; see `g1t_contracts::billing` for
18	//! the methods and their arguments.
19
20	mod accounts;
21	mod budget;
22	mod charged;
23	mod cards;
24	mod closing;
25	mod compute;
26	mod costs;
27	mod cycle;
28	mod margin;
29	mod platform;
30	mod pricing;
31	mod report;
32	mod details;
33	mod credits;
34	mod grants;
35	mod overages;
36	mod requests;
37	mod storage;
38	mod invoices;
39	mod sales;
40	mod statement;
41	mod webhooks;
42	mod features;
43	mod keeper;
44	mod limits;
45	mod rename;
46	mod reset;
47	mod retention;
48	mod stripe;
49	mod stripe_sync;
50	mod subscriptions;
51	mod tax;
52
53	use g1t_contracts::billing::*;
54	use g1t_contracts::time::rfc3339;
55	use g1t_contracts::{FailureCode, Outcome, new_id};
56	use g1t_kit::{args, now_ms, reply, rpc_method};
57	use serde::Deserialize;
58	use worker::wasm_bindgen::JsValue;
59	use worker::{Context, D1Database, Env, MessageBatch, MessageExt, Request, Response, Result, ScheduleContext, ScheduledEvent, event};
60	use futures_util::future::{try_join, try_join5};
61
62	use stripe::Stripe;
63
64	/// Prepaying: $25 at the least; by card up to $10,000 at a time, and by
65	/// bank transfer from $1,000 to $100,000.
66	const MIN_TOP_UP_CENTS: u32 = 2_500;
67	const MAX_TOP_UP_CENTS: u32 = 1_000_000;
68	const MIN_BANK_TRANSFER_CENTS: u32 = 100_000;
69	const MAX_BANK_TRANSFER_CENTS: u32 = 10_000_000;
70	const LEDGER_PAGE: u32 = 100;
71	/// What a cost g1t trusts (the price book's) is charged at: plus the
72	/// margin, rounded up to a whole millionth.
73	pub fn margin_on(cost_micros: i64, margin_percent: u32) -> i64 {
74	let cost = i128::from(cost_micros.max(0));
75	let charge = (cost * i128::from(100 + margin_percent) + 99) / 100;
76	i64::try_from(charge).unwrap_or(i64::MAX)
77	}
78
79	fn optional(value: Option<&str>) -> JsValue {
80	value.map_or(JsValue::NULL, JsValue::from)
81	}
82
83	#[derive(Deserialize)]
84	struct AccountRow {
85	balance_micros: i64,
86	customer_id: Option<String>,
87	}
88
89	#[derive(Deserialize)]
90	struct LedgerRow {
91	id: String,
92	kind: EntryKind,
93	amount_micros: i64,
94	description: String,
95	repo: Option<String>,
96	number: Option<u32>,
97	task: Option<String>,
98	model: Option<String>,
99	created_by: Option<String>,
100	created_at: String,
101	billed_to: Option<String>,
102	#[serde(default)]
103	workspace: Option<String>,
104	#[serde(default)]
105	credit_micros: Option<i64>,
106	#[serde(default)]
107	trial_micros: Option<i64>,
108	#[serde(default)]
109	oss_micros: Option<i64>,
110	#[serde(default)]
111	given_micros: Option<i64>,
112	#[serde(default)]
113	credit_kind: Option<String>,
114	#[serde(default)]
115	discount_micros: Option<i64>,
116	}
117
118	impl From<LedgerRow> for LedgerEntry {
119	fn from(row: LedgerRow) -> Self {
120	LedgerEntry {
121	id: row.id,
122	kind: row.kind,
123	amount_micros: row.amount_micros,
124	description: row.description,
125	repo: row.repo,
126	number: row.number,
127	task: row.task,
128	model: row.model,
129	billed_to: row.billed_to.unwrap_or_else(\|\| "g1t".to_owned()),
130	created_by: row.created_by,
131	created_at: row.created_at,
132	workspace: row.workspace,
133	credit_micros: row.credit_micros.unwrap_or(0),
134	trial_micros: row.trial_micros.unwrap_or(0),
135	oss_micros: row.oss_micros.unwrap_or(0),
136	given_micros: row.given_micros.unwrap_or(0),
137	credit_kind: row.credit_kind.as_deref().and_then(CreditKind::parse),
138	discount_micros: row.discount_micros.unwrap_or(0),
139	}
140	}
141	}
142
143	#[derive(Deserialize)]
144	struct CheckoutRow {
145	workspace: String,
146	created_by: String,
147	#[serde(default)]
148	fee_cents: Option<u32>,
149	}
150
151	/// A payment page started, as `checkouts` keeps it.
152	pub(crate) struct NewCheckout<'a> {
153	pub id: &'a str,
154	pub workspace: &'a str,
155	/// What it pays for, in cents: the prepayment or the plan's price; 0
156	/// for a card check.
157	pub amount_cents: u32,
158	/// The card fee on top.
159	pub fee_cents: u32,
160	pub created_by: &'a str,
161	/// None for a prepayment; `plan`, `security` or `card_check` otherwise
162	/// (`ai_credit` on pages from before AI credit was retired).
163	pub feature: Option<&'a str>,
164	}
165
166	/// The one insert every payment page goes through, every column named, so
167	/// a column added to `checkouts` with no default is caught by
168	/// `tests::every_checkout_insert_fills_the_table` rather than by a 500.
169	pub(crate) const CHECKOUT_INSERT: &str = "INSERT INTO checkouts (id, workspace, amount_cents, fee_cents, created_by, created_at, feature, status)
170	VALUES (?, ?, ?, ?, ?, ?, ?, 'open')";
171
172	impl Billing {
173	pub(crate) async fn record_checkout(&self, c: &NewCheckout<'_>) -> Result<()> {
174	self.db
175	.prepare(CHECKOUT_INSERT)
176	.bind(&[
177	c.id.into(),
178	c.workspace.into(),
179	c.amount_cents.into(),
180	c.fee_cents.into(),
181	c.created_by.into(),
182	rfc3339(now_ms()).into(),
183	optional(c.feature),
184	])?
185	.run()
186	.await?;
187	Ok(())
188	}
189	}
190
191	/// A row an `UPDATE … RETURNING` touched.
192	#[derive(Deserialize)]
193	struct Touched {
194	#[allow(dead_code)]
195	id: String,
196	}
197
198	struct Billing {
199	db: D1Database,
200	/// Absent when no card processor is configured.
201	stripe: Option<Stripe>,
202	/// The destination's signing secret from Stripe (`STRIPE_WEBHOOK_SECRET`);
203	/// without it no event is believed.
204	webhook_secret: Option<String>,
205	margin_percent: u32,
206	/// While g1t is being built out, nothing is charged (`FREE_WHILE_BUILDING`).
207	free: bool,
208	/// Whether new workspaces get trial credit (`TRIAL_WORKSPACE_MICROS`
209	/// and `TRIAL_MONTHLY_POOL_MICROS` both above zero).
210	trials_on: bool,
211	/// The plans' and pools' numbers; see `credits`.
212	plans: credits::Config,
213	/// The repos service: which repositories are public, what private ones
214	/// hold, and their git operations. Absent where it is not bound.
215	repos: Option<worker::Fetcher>,
216	/// The packages service: what each workspace's packages hold.
217	packages: Option<worker::Fetcher>,
218	/// The identity service, which emails owners. Absent where it is not
219	/// bound.
220	identity: Option<worker::Fetcher>,
221	/// How far unpaid usage may go; see `limits`.
222	ceilings: limits::Ceilings,
223	/// The caps on what g1t pays for itself; see `budget`.
224	caps: budget::Caps,
225	/// The worker's bindings, for emailing staff (`EMAIL`).
226	env: Env,
227	}
228
229	impl Billing {
230	fn status(&self) -> Status {
231	Status {
232	enabled: self.stripe.is_some(),
233	live: self.stripe.as_ref().is_some_and(Stripe::live),
234	free: self.free,
235	}
236	}
237
238	async fn row(&self, workspace: &str) -> Result<Option<AccountRow>> {
239	self.db
240	.prepare("SELECT balance_micros, customer_id FROM accounts WHERE workspace = ?")
241	.bind(&[workspace.into()])?
242	.first::<AccountRow>(None)
243	.await
244	}
245
246	async fn standing(&self, workspace: &str) -> Result<Account> {
247	// The card as last synced (stripe_sync.rs), not asked of Stripe.
248	let (row, card) = try_join(self.row(workspace), self.saved_card(workspace)).await?;
249	Ok(Account {
250	workspace: workspace.to_owned(),
251	balance_micros: row.map_or(0, \|row\| row.balance_micros),
252	status: self.status(),
253	margin_percent: self.margin_percent,
254	card,
255	})
256	}
257
258	/// The workspace's customer at Stripe, made the first time one is needed.
259	pub(crate) async fn customer_for(&self, workspace: &str) -> Result<String> {
260	let Some(stripe) = &self.stripe else {
261	return Err(worker::Error::RustError("payments are not set up".into()));
262	};
263	if let Some(customer) = self.row(workspace).await?.and_then(\|row\| row.customer_id) {
264	return Ok(customer);
265	}
266	let customer = stripe.create_customer(workspace).await?;
267	self.db
268	.prepare(
269	"INSERT INTO accounts (workspace, balance_micros, customer_id, created_at) VALUES (?1, 0, ?2, ?3)
270	ON CONFLICT (workspace) DO UPDATE SET customer_id = ?2",
271	)
272	.bind(&[workspace.into(), customer.as_str().into(), rfc3339(now_ms()).into()])?
273	.run()
274	.await?;
275	Ok(customer)
276	}
277
278	/// Stripe's hosted billing page for the workspace. Owners only.
279	async fn billing_portal(&self, a: BillingPortalArgs) -> Result<Outcome<Checkout>> {
280	let workspace = a.workspace.to_lowercase();
281	if !a.actor.manages_billing(&workspace) {
282	return Ok(Outcome::fail(FailureCode::Forbidden, "Only an owner or a billing manager can manage the workspace's billing."));
283	}
284	let Some(stripe) = &self.stripe else {
285	return Ok(Outcome::fail(FailureCode::Conflict, "Payments are not set up on this g1t."));
286	};
287	let customer = match self.customer_for(&workspace).await {
288	Ok(customer) => customer,
289	Err(error) => return Ok(Outcome::fail(FailureCode::Conflict, crate::stripe::friendly(&error))),
290	};
291	match stripe.portal_session(&customer, &a.return_url).await {
292	Ok(url) => Ok(Outcome::Ok(Checkout { url })),
293	Err(error) if stripe::is_missing(&error) => {
294	// The customer was removed at Stripe: a new one next time.
295	self.forget_customer(&workspace).await?;
296	Ok(Outcome::fail(FailureCode::Conflict, "Stripe no longer had this workspace's customer. Try again."))
297	}
298	Err(error) => Ok(Outcome::fail(FailureCode::Conflict, crate::stripe::friendly(&error))),
299	}
300	}
301
302	/// Adds a ledger entry and moves the balance by the same amount, as
303	/// one write.
304	#[allow(clippy::too_many_arguments)]
305	async fn enter(
306	&self,
307	workspace: &str,
308	kind: EntryKind,
309	amount_micros: i64,
310	description: &str,
311	reference: &str,
312	cost_micros: Option<i64>,
313	created_by: Option<&str>,
314	customer: Option<&str>,
315	) -> Result<()> {
316	let now = now_ms();
317	let timestamp = rfc3339(now);
318	let kind = match kind {
319	EntryKind::TopUp => "top_up",
320	EntryKind::Usage => "usage",
321	};
322	self.db
323	.batch(vec![
324	self.db
325	.prepare(
326	"INSERT INTO ledger
327	(id, workspace, kind, amount_micros, description, cost_micros, reference, created_by, created_at, billed_to)
328	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'g1t')",
329	)
330	.bind(&[
331	new_id("led", now).into(),
332	workspace.into(),
333	kind.into(),
334	// D1 takes numbers as doubles, which hold every
335	// amount this service will see exactly.
336	(amount_micros as f64).into(),
337	description.into(),
338	cost_micros.map_or(JsValue::NULL, \|cost\| (cost as f64).into()),
339	reference.into(),
340	optional(created_by),
341	timestamp.as_str().into(),
342	])?,
343	self.db
344	.prepare(
345	"INSERT INTO accounts (workspace, balance_micros, customer_id, created_at)
346	VALUES (?1, ?2, ?3, ?4)
347	ON CONFLICT (workspace) DO UPDATE SET
348	balance_micros = balance_micros + ?2,
349	customer_id = COALESCE(?3, customer_id)",
350	)
351	.bind(&[
352	workspace.into(),
353	(amount_micros as f64).into(),
354	optional(customer),
355	timestamp.as_str().into(),
356	])?,
357	])
358	.await?;
359	// Money in clears a card declined at the limit. A refund or a lost
360	// dispute is a top-up of less than nothing: money out, which never
361	// clears it.
362	if kind == "top_up" && amount_micros > 0 {
363	self.db
364	.prepare("UPDATE limits SET autopay_failed_at = NULL, autopay_error = NULL WHERE workspace = ?")
365	.bind(&[workspace.into()])?
366	.run()
367	.await?;
368	}
369	Ok(())
370	}
371
372	async fn account(&self, a: AccountArgs) -> Result<Outcome<Account>> {
373	let workspace = a.workspace.to_lowercase();
374	if !a.viewer.is_some_and(\|viewer\| viewer.is_member(&workspace)) {
375	return Ok(members_only());
376	}
377	Ok(Outcome::Ok(self.standing(&workspace).await?))
378	}
379
380	async fn ledger(&self, a: AccountArgs) -> Result<Outcome<Vec<LedgerEntry>>> {
381	let workspace = a.workspace.to_lowercase();
382	if !a.viewer.is_some_and(\|viewer\| viewer.is_member(&workspace)) {
383	return Ok(members_only());
384	}
385	let rows = self
386	.db
387	.prepare("SELECT * FROM ledger WHERE workspace = ? ORDER BY id DESC LIMIT ?")
388	.bind(&[workspace.into(), LEDGER_PAGE.into()])?
389	.all()
390	.await?
391	.results::<LedgerRow>()?;
392	Ok(Outcome::Ok(
393	rows.into_iter().map(LedgerEntry::from).collect(),
394	))
395	}
396
397	async fn usage(&self, a: UsageArgs) -> Result<Outcome<Usage>> {
398	let workspace = a.workspace.to_lowercase();
399	if !a.viewer.is_some_and(\|viewer\| viewer.is_member(&workspace)) {
400	return Ok(members_only());
401	}
402	#[derive(serde::Deserialize)]
403	struct SliceRow {
404	key: Option<String>,
405	micros: Option<i64>,
406	runs: Option<u32>,
407	}
408	// While g1t is free, what was used at cost is what there is to show.
409	// With a discount, usage at its price, so a 100% discount still
410	// shows what the workspace would pay.
411	// Every slice measures usage at price (statement::PRICE_SQL), the one
412	// figure every page shows as usage, whatever paid for it.
413	let discount_percent = self.terms_of(&workspace).await?.percent_off();
414	let measure = if self.free { "COALESCE(cost_micros, 0)" } else { crate::statement::PRICE_SQL };
415	let slices = \|key: &str, limit: u32\| {
416	format!(
417	"SELECT {key} AS key, SUM({measure}) AS micros, COUNT(*) AS runs FROM ledger
418	WHERE workspace = ?1 AND kind = 'usage' AND created_at >= ?2
419	GROUP BY 1 ORDER BY micros DESC LIMIT {limit}"
420	)
421	};
422	let query = \|sql: String\| {
423	let db = &self.db;
424	let workspace = workspace.clone();
425	let since = a.since.clone();
426	async move {
427	let rows = db
428	.prepare(sql)
429	.bind(&[workspace.into(), since.into()])?
430	.all()
431	.await?
432	.results::<SliceRow>()?;
433	Ok::<Vec<UsageSlice>, worker::Error>(
434	rows.into_iter()
435	.map(\|row\| UsageSlice {
436	key: row.key.unwrap_or_else(\|\| "other".to_owned()),
437	micros: row.micros.unwrap_or_default(),
438	runs: row.runs.unwrap_or_default(),
439	})
440	.collect(),
441	)
442	}
443	};
444	#[derive(serde::Deserialize)]
445	struct Totals {
446	spent: Option<i64>,
447	cost: Option<i64>,
448	provider: Option<i64>,
449	runs: Option<u32>,
450	added: Option<i64>,
451	discount: Option<i64>,
452	covered: Option<i64>,
453	price: Option<i64>,
454	}
455	let totals = async {
456	self.db
457	.prepare(format!(
458	"SELECT
459	-SUM(CASE WHEN kind = 'usage' THEN amount_micros END) AS spent,
460	SUM(CASE WHEN kind = 'usage' THEN COALESCE(credit_micros, 0) + COALESCE(trial_micros, 0)
461	+ COALESCE(oss_micros, 0) + COALESCE(given_micros, 0) END) AS covered,
462	SUM(CASE WHEN kind = 'usage' THEN {price} END) AS price,",
463	price = crate::statement::PRICE_SQL
464	) + "
465	SUM(CASE WHEN kind = 'usage' AND COALESCE(billed_to, 'g1t') = 'g1t' THEN cost_micros END) AS cost,
466	SUM(CASE WHEN kind = 'usage' AND billed_to = 'workspace' THEN cost_micros END) AS provider,
467	SUM(CASE WHEN kind = 'usage' AND " + crate::statement::RUN_SQL + " THEN 1 ELSE 0 END) AS runs,
468	SUM(CASE WHEN kind = 'top_up' THEN amount_micros END) AS added,
469	SUM(CASE WHEN kind = 'usage' THEN discount_micros END) AS discount
470	FROM ledger WHERE workspace = ?1 AND created_at >= ?2",
471	)
472	.bind(&[workspace.as_str().into(), a.since.as_str().into()])?
473	.first::<Totals>(None)
474	.await
475	};
476	// The totals and the five slices read the same rows independently,
477	// so they go to D1 at once: one round trip of waiting, not six.
478	let (totals, (by_day, by_task, by_repo, by_pull, by_model)) = try_join(
479	totals,
480	try_join5(
481	query(slices("substr(created_at, 1, 10) \|\| '/' \|\| COALESCE(task, 'other')", 400)),
482	query(slices("task", 20)),
483	query(slices("repo", 20)),
484	query(slices("repo \|\| '#' \|\| number", 10)),
485	query(slices("model", 10)),
486	),
487	)
488	.await?;
489	let totals = totals.unwrap_or(Totals {
490	spent: None,
491	cost: None,
492	provider: None,
493	runs: None,
494	added: None,
495	discount: None,
496	covered: None,
497	price: None,
498	});
499	Ok(Outcome::Ok(Usage {
500	spent_micros: totals.spent.unwrap_or_default(),
501	// What the included usage, the trial, a pool or g1t paid, as the
502	// entries record it.
503	covered_micros: totals.covered.unwrap_or_default(),
504	price_micros: if self.free { totals.cost.unwrap_or_default() } else { totals.price.unwrap_or_default() },
505	cost_micros: totals.cost.unwrap_or_default(),
506	provider_micros: totals.provider.unwrap_or_default(),
507	used_micros: totals.cost.unwrap_or_default() + totals.provider.unwrap_or_default(),
508	free: self.free,
509	discount_micros: totals.discount.unwrap_or_default(),
510	discount_percent: (discount_percent > 0).then_some(discount_percent),
511	runs: totals.runs.unwrap_or_default(),
512	added_micros: totals.added.unwrap_or_default(),
513	by_day,
514	by_task,
515	by_repo,
516	by_pull,
517	by_model,
518	since: a.since,
519	}))
520	}
521
522	/// `checkout`: prepays usage, by card or (from $1,000) bank transfer.
523	async fn checkout(&self, a: CheckoutArgs) -> Result<Outcome<Checkout>> {
524	let workspace = a.workspace.to_lowercase();
525	if !a.actor.manages_billing(&workspace) {
526	return Ok(Outcome::fail(
527	FailureCode::Forbidden,
528	"Only an owner or a billing manager can prepay for a workspace.",
529	));
530	}
531	let Some(stripe) = &self.stripe else {
532	return Ok(Outcome::fail(
533	FailureCode::Conflict,
534	"Payments are not set up on this g1t yet.",
535	));
536	};
537	let bank_transfer = a.method.as_deref() == Some("bank_transfer");
538	if let Err(why) = prepay_amount(a.amount_cents, bank_transfer) {
539	return Ok(Outcome::fail(FailureCode::Invalid, why));
540	}
541	// A bank transfer needs a customer for its account details.
542	let customer = if bank_transfer {
543	match self.customer_for(&workspace).await {
544	Ok(customer) => Some(customer),
545	Err(error) => return Ok(Outcome::fail(FailureCode::Conflict, stripe::friendly(&error))),
546	}
547	} else {
548	self.row(&workspace).await?.and_then(\|row\| row.customer_id)
549	};
550	// By card, Stripe's fee is its own line; a bank transfer has none.
551	let fee = if bank_transfer { 0 } else { self.card_fee_on(i64::from(a.amount_cents)).await? as u32 };
552	let started = match stripe.start_checkout(&workspace, a.amount_cents, fee, customer.as_deref(), &a.return_url, bank_transfer).await {
553	// A customer saved under another Stripe account: start afresh.
554	Err(error) if customer.is_some() && stripe::is_missing(&error) => {
555	self.forget_customer(&workspace).await?;
556	let customer = if bank_transfer { self.customer_for(&workspace).await.ok() } else { None };
557	stripe.start_checkout(&workspace, a.amount_cents, fee, customer.as_deref(), &a.return_url, bank_transfer).await
558	}
559	other => other,
560	};
561	self.page_opened(started, &workspace, a.amount_cents, fee, &a.actor.username, None).await
562	}
563
564	/// A payment page Stripe started (or refused), recorded in `checkouts`
565	/// and handed back; any failure as a sentence for the page, never a
566	/// raw error.
567	pub(crate) async fn page_opened(
568	&self,
569	started: Result<stripe::Session>,
570	workspace: &str,
571	amount_cents: u32,
572	fee_cents: u32,
573	created_by: &str,
574	feature: Option<&str>,
575	) -> Result<Outcome<Checkout>> {
576	let session = match started {
577	Ok(session) => session,
578	Err(error) => {
579	worker::console_error!("{workspace}: Stripe refused a payment page ({}): {error}", feature.unwrap_or("prepay"));
580	return Ok(Outcome::fail(FailureCode::Conflict, stripe::friendly(&error)));
581	}
582	};
583	let Some(url) = session.url.clone() else {
584	return Ok(Outcome::fail(FailureCode::Conflict, "Stripe returned no payment page. Try again in a minute."));
585	};
586	let recorded = self
587	.record_checkout(&NewCheckout { id: &session.id, workspace, amount_cents, fee_cents, created_by, feature })
588	.await;
589	if let Err(error) = recorded {
590	worker::console_error!("{workspace}: a payment page could not be recorded: {error}");
591	return Ok(Outcome::fail(FailureCode::Conflict, "g1t could not keep track of the payment page. Nothing was charged; try again."));
592	}
593	Ok(Outcome::Ok(Checkout { url }))
594	}
595
596	/// Credits a payment if the processor says it was made and it has not
597	/// been credited before. The amount credited is what the processor
598	/// says was paid, not what anyone here remembers asking for.
599	async fn confirm(&self, a: ConfirmArgs) -> Result<Outcome<Account>> {
600	let workspace = a.workspace.to_lowercase();
601	if !a.viewer.is_some_and(\|viewer\| viewer.is_member(&workspace)) {
602	return Ok(members_only());
603	}
604	let (Some(stripe), Some(checkout)) = (
605	&self.stripe,
606	self.db
607	.prepare(
608	// A prepayment only: a plan's or a card check's page is
609	// settled where it was started, never credited as money
610	// paid in advance.
611	"SELECT workspace, created_by, fee_cents FROM checkouts
612	WHERE id = ? AND workspace = ? AND status = 'open' AND feature IS NULL",
613	)
614	.bind(&[a.session.as_str().into(), workspace.as_str().into()])?
615	.first::<CheckoutRow>(None)
616	.await?,
617	) else {
618	// Unknown, someone else's, or already credited: nothing to do.
619	return Ok(Outcome::Ok(self.standing(&workspace).await?));
620	};
621	let session = match stripe.session(&a.session).await {
622	Ok(session) => session,
623	Err(error) => return Ok(Outcome::fail(FailureCode::Conflict, crate::stripe::friendly(&error))),
624	};
625	if session.payment_status == "paid" && session.amount_total.is_some() {
626	// Only whoever flips it from open to paid enters the credit.
627	let claimed = self
628	.db
629	.prepare(
630	"UPDATE checkouts SET status = 'paid' WHERE id = ? AND status = 'open'
631	RETURNING id",
632	)
633	.bind(&[a.session.as_str().into()])?
634	.first::<Touched>(None)
635	.await?;
636	if claimed.is_some() {
637	self.credit_prepayment(&checkout.workspace, &session, checkout.fee_cents.unwrap_or(0), &checkout.created_by).await?;
638	}
639	}
640	Ok(Outcome::Ok(self.standing(&workspace).await?))
641	}
642
643	/// Enters a paid prepayment page: what its lines came to before tax,
644	/// less the card fee, as credit; its tax and fee kept apart. Once the
645	/// page is claimed `open → paid`. The credit in cents.
646	pub(crate) async fn credit_prepayment(&self, workspace: &str, session: &stripe::Session, fee_cents: u32, by: &str) -> Result<i64> {
647	let fee = i64::from(fee_cents);
648	let cents = tax::prepay_credit_cents(session.before_tax_cents(), fee);
649	let tax_cents = session.tax_cents();
650	let description = if tax_cents > 0 \|\| fee > 0 {
651	format!(
652	"Paid in advance (tax {} and card fee {} paid with it)",
653	features::cents(tax_cents * 10_000),
654	features::cents(fee * 10_000)
655	)
656	} else {
657	"Paid in advance".to_owned()
658	};
659	self.enter(
660	workspace,
661	EntryKind::TopUp,
662	cents * MICROS_PER_DOLLAR / 100,
663	&description,
664	&session.id,
665	None,
666	Some(by),
667	session.customer.as_deref(),
668	)
669	.await?;
670	self.record_extras(workspace, &session.id, session.payment_intent.as_deref(), tax::Extras { tax_cents, fee_cents: fee }, None)
671	.await?;
672	Ok(cents)
673	}
674
675	/// Drops a saved customer the card processor no longer knows.
676	pub(crate) async fn forget_customer(&self, workspace: &str) -> Result<()> {
677	self.db
678	.prepare(
679	"UPDATE accounts SET customer_id = NULL, card_brand = NULL, card_last4 = NULL, card_exp_month = NULL,
680	card_exp_year = NULL, card_synced_at = NULL WHERE workspace = ?",
681	)
682	.bind(&[workspace.into()])?
683	.run()
684	.await?;
685	Ok(())
686	}
687	}
688
689	impl Billing {
690	/// Records how long a sandbox ran, with its cost and its charge: every
691	/// second, from the first, at the price book's price; on its own CPU
692	/// when it reports it. Settles its reservation, if it names one.
693	async fn record_sandbox(&self, a: RecordSandboxArgs) -> Result<Outcome<bool>> {
694	// Self-hosted time is recorded wherever g1t runs, for its minutes;
695	// anything else only where there is a bill to put it on.
696	if (self.stripe.is_none() && !a.self_hosted) \|\| a.seconds == 0 {
697	return Ok(Outcome::Ok(false));
698	}
699	let workspace = a.workspace.to_lowercase();
700	let seen = self
701	.db
702	.prepare("SELECT id FROM ledger WHERE reference = ?")
703	.bind(&[a.reference.as_str().into()])?
704	.first::<Touched>(None)
705	.await?;
706	if seen.is_some() {
707	return Ok(Outcome::Ok(false));
708	}
709	let now = now_ms();
710	let timestamp = rfc3339(now);
711	let seconds = i64::from(a.seconds);
712	// On the workspace's own machine: its minutes go on usage, at $0,
713	// and whatever was reserved for it is given back.
714	if a.self_hosted {
715	let description = format!("{}: {} of self-hosted runner time, $0", a.description, duration(seconds));
716	self.db
717	.prepare(
718	"INSERT INTO ledger
719	(id, workspace, kind, amount_micros, description, repo, task,
720	cost_micros, reference, created_at, billed_to, credit_micros, trial_micros, oss_micros, given_micros, quantity)
721	VALUES (?, ?, 'usage', 0, ?, ?, 'self_hosted', 0, ?, ?, 'workspace', 0, 0, 0, 0, ?)",
722	)
723	.bind(&[
724	new_id("led", now).into(),
725	workspace.as_str().into(),
726	description.as_str().into(),
727	optional(a.repo.as_deref()),
728	a.reference.as_str().into(),
729	timestamp.as_str().into(),
730	(seconds as f64).into(),
731	])?
732	.run()
733	.await?;
734	if let Some(reservation) = &a.reservation_id {
735	self.settle_reservation(SettleArgs { reservation_id: reservation.clone(), actual_micros: 0 }).await?;
736	}
737	return Ok(Outcome::Ok(true));
738	}
739	// From the price book, which follows what Cloudflare bills g1t. A
740	// sandbox is the same container as a build, so without a row it is
741	// a build second's cost plus the margin.
742	let (cost_per_second, _) = self.price("sandbox_second").await?.unwrap_or_else(\|\| {
743	let cost = deployment_costs::MICROS_PER_BUILD_SECOND as f64;
744	(cost, Price::price_for(cost, self.margin_percent))
745	});
746	// The machine it ran on, from the catalogue by its `runs-on` label;
747	// the standard one otherwise. What it cost g1t: a larger machine's
748	// memory and disk cost more each second, and without its own CPU it
749	// is costed at its vCPUs as busy as the standard machine's.
750	let instance = a.instance.as_deref().and_then(g1t_contracts::actions::instance_named).unwrap_or(g1t_contracts::actions::STANDARD_INSTANCE);
751	let base_scale = keeper::base_scale(instance.memory_gib, instance.disk_gb);
752	let cost_scale = keeper::busy_scale(&instance);
753	let vcpus = instance.vcpu.max(4.0);
754	// Its own CPU when the sandbox reports it; otherwise the average.
755	let parts = match a.cpu_seconds.filter(\|cpu\| cpu.is_finite() && *cpu >= 0.0) {
756	Some(cpu) => match (self.price("sandbox_base_second").await?, self.price("sandbox_cpu_second").await?) {
757	(Some((base, _)), Some((vcpu, _))) => {
758	Some((keeper::run_cost(seconds, cpu.min(seconds as f64 * vcpus), base * base_scale, vcpu), cpu))
759	}
760	_ => None,
761	},
762	None => None,
763	};
764	let cost = parts.map_or(seconds as f64 * cost_per_second * cost_scale, \|(cost, _)\| cost).ceil() as i64;
765	// Charged by the job: its minutes, rounded up, at the machine's rate
766	// (never under what it cost g1t).
767	let (minutes, price) = keeper::runner_charge(seconds, &instance, cost);
768	let (charge, terms_note, discount) = self.charged(&workspace, price).await?;
769	let eligible = credits::eligible_for(a.kind, a.repo.as_deref());
770	let drawn = self.draw(&workspace, charge, &credits::month_of(&timestamp), &eligible).await?;
771	let charge = charge - drawn.total();
772	let description = format!(
773	"{}: {}, {} on the {} machine{terms_note}{}",
774	a.description,
775	duration(seconds),
776	keeper::minutes_text(minutes),
777	instance.machine,
778	drawn.note()
779	);
780	// The price versions it was charged at (pricing.rs).
781	let meters: &[&str] = if parts.is_some() { &["sandbox_base_second", "sandbox_cpu_second"] } else { &["sandbox_second"] };
782	let mut versions = Vec::new();
783	for meter in meters {
784	versions.extend(self.version_now(meter).await?);
785	}
786	let price_version = versions.join(",");
787	self.db
788	.batch(vec![
789	self.db
790	.prepare(
791	"INSERT INTO ledger
792	(id, workspace, kind, amount_micros, description, repo, task,
793	cost_micros, reference, created_at, billed_to, credit_micros, trial_micros, oss_micros, given_micros, price_version, quantity, compute, machine)
794	VALUES (?, ?, 'usage', ?, ?, ?, 'sandbox', ?, ?, ?, 'g1t', ?, ?, ?, ?, ?, ?, ?, ?)",
795	)
796	.bind(&[
797	new_id("led", now).into(),
798	workspace.as_str().into(),
799	(-(charge as f64)).into(),
800	description.as_str().into(),
801	optional(a.repo.as_deref()),
802	(cost as f64).into(),
803	a.reference.as_str().into(),
804	timestamp.as_str().into(),
805	(drawn.credit as f64).into(),
806	(drawn.trial as f64).into(),
807	(drawn.oss as f64).into(),
808	(drawn.given as f64).into(),
809	optional(Some(price_version.as_str()).filter(\|v\| !v.is_empty())),
810	// Runner minutes, as charged; lines with no machine
811	// (before minutes) hold seconds.
812	(minutes as f64).into(),
813	optional(a.kind.map(ComputeKind::as_str)),
814	instance.machine.into(),
815	])?,
816	self.db
817	.prepare(
818	"INSERT INTO accounts (workspace, balance_micros, created_at)
819	VALUES (?1, ?2, ?3)
820	ON CONFLICT (workspace) DO UPDATE SET balance_micros = balance_micros + ?2",
821	)
822	.bind(&[
823	workspace.as_str().into(),
824	(-(charge as f64)).into(),
825	timestamp.as_str().into(),
826	])?,
827	])
828	.await?;
829	self.record_discount(&a.reference, discount).await?;
830	self.count_spend(&workspace, cost, charge, &drawn).await;
831	if let Some(reservation) = &a.reservation_id {
832	self.settle_reservation(SettleArgs { reservation_id: reservation.clone(), actual_micros: cost }).await?;
833	}
834	Ok(Outcome::Ok(true))
835	}
836	}
837
838	/// Whether an amount may be prepaid: $25 at the least by card, and from
839	/// $1,000 by bank transfer.
840	pub(crate) fn prepay_amount(cents: u32, bank_transfer: bool) -> std::result::Result<(), String> {
841	let (min, max) = if bank_transfer { (MIN_BANK_TRANSFER_CENTS, MAX_BANK_TRANSFER_CENTS) } else { (MIN_TOP_UP_CENTS, MAX_TOP_UP_CENTS) };
842	if (min..=max).contains(&cents) {
843	return Ok(());
844	}
845	Err(if bank_transfer {
846	format!("Prepay between ${} and ${} by bank transfer.", min / 100, group(max / 100))
847	} else {
848	format!("Prepay between ${} and ${} by card; from $1,000, a bank transfer works too.", min / 100, group(max / 100))
849	})
850	}
851
852	/// `10,000` for 10000.
853	fn group(n: u32) -> String {
854	let digits = n.to_string();
855	let mut out = String::new();
856	for (i, c) in digits.chars().enumerate() {
857	if i > 0 && (digits.len() - i).is_multiple_of(3) {
858	out.push(',');
859	}
860	out.push(c);
861	}
862	out
863	}
864
865	#[cfg(test)]
866	/// What `seconds` of sandbox time are charged at `price_per_second`, in
867	/// millionths of a dollar: every second, rounded up to the next millionth.
868	fn sandbox_charge(seconds: i64, price_per_second: f64) -> i64 {
869	(seconds as f64 * price_per_second).ceil() as i64
870	}
871
872	/// `1h 2m`, `3m 12s` or `40s`.
873	fn duration(seconds: i64) -> String {
874	let (h, m, s) = (seconds / 3600, seconds % 3600 / 60, seconds % 60);
875	if h > 0 {
876	format!("{h}h {m}m")
877	} else if m > 0 {
878	format!("{m}m {s}s")
879	} else {
880	format!("{s}s")
881	}
882	}
883
884	impl Billing {
885	/// What a workspace is charged for something that would be `base`:
886	/// nothing while g1t is free, or as its account's terms say. With a
887	/// note for the statement when it differs.
888	/// A charge at cost plus the margin (`base`) on the account's terms:
889	/// what is charged, the note for the statement, and what a discount
890	/// gave away below `base`. That last is written on the entry
891	/// (`record_discount`) so the reconciliation counts it as given, never
892	/// as margin lost: a sold charge is worth at least its cost plus the
893	/// margin.
894	pub(crate) async fn charged(&self, workspace: &str, base: i64) -> Result<(i64, String, i64)> {
895	if self.free {
896	return Ok((0, " (free while g1t is being built out)".to_owned(), 0));
897	}
898	let terms = self.terms_of(workspace).await?;
899	let (charge, discount) = terms.discounted(base);
900	let note = match terms.discount_label() {
901	Some(label) if base > 0 => format!(" ({label})"),
902	_ => String::new(),
903	};
904	Ok((charge, note, discount))
905	}
906
907	/// What a discount gave away on an entry, below cost plus the margin
908	/// (less than nothing on a correction down).
909	pub(crate) async fn record_discount(&self, reference: &str, micros: i64) -> Result<()> {
910	if micros == 0 {
911	return Ok(());
912	}
913	self.db
914	.prepare("UPDATE ledger SET discount_micros = ? WHERE reference = ?")
915	.bind(&[(micros as f64).into(), reference.into()])?
916	.run()
917	.await?;
918	Ok(())
919	}
920	}
921
922	fn members_only<T>() -> Outcome<T> {
923	Outcome::fail(
924	FailureCode::Forbidden,
925	"Only members can see a workspace's billing.",
926	)
927	}
928
929	impl Billing {
930	fn from_env(env: &Env) -> Result<Self> {
931	Ok(Billing {
932	db: env.d1("DB")?,
933	stripe: env
934	.secret("STRIPE_SECRET_KEY")
935	.ok()
936	.map(\|key\| key.to_string())
937	.filter(\|key\| !key.is_empty())
938	.map(Stripe::new),
939	webhook_secret: env
940	.secret("STRIPE_WEBHOOK_SECRET")
941	.ok()
942	.map(\|secret\| secret.to_string().trim().to_owned())
943	.filter(\|secret\| !secret.is_empty()),
944	margin_percent: env
945	.var("MARGIN_PERCENT")
946	.ok()
947	.and_then(\|percent\| percent.to_string().parse().ok())
948	.unwrap_or(20),
949	free: env.var("FREE_WHILE_BUILDING").is_ok_and(\|v\| v.to_string() == "true"),
950	ceilings: limits::Ceilings::from_env(env),
951	trials_on: {
952	let plans = credits::Config::from_env(env);
953	plans.trial_workspace_micros > 0 && plans.trial_monthly_pool_micros > 0
954	},
955	plans: credits::Config::from_env(env),
956	repos: env.service("REPOS").ok(),
957	packages: env.service("PACKAGES").ok(),
958	identity: env.service("IDENTITY").ok(),
959	caps: budget::Caps::from_env(env),
960	env: env.clone(),
961	})
962	}
963	}
964
965	/// The fifteen-minute steps, each on its own: one failing never skips the
966	/// rest.
967	async fn tick(billing: &Billing, env: &Env) {
968	// Stripe events billing never received, handled now.
969	match billing.replay_events().await {
970	Ok(done) => worker::console_log!("stripe replay: {done}"),
971	Err(error) => worker::console_error!("replaying Stripe events failed: {error}"),
972	}
973	// This month's users, read once for autopay and the warnings.
974	let users = if billing.stripe.is_some() {
975	match billing.month_users().await {
976	Ok(users) => Some(users),
977	Err(error) => {
978	worker::console_error!("reading this month's users failed: {error}");
979	None
980	}
981	}
982	} else {
983	None
984	};
985	if let Some(users) = &users
986	&& let Err(error) = billing.autopay(users).await {
987	worker::console_error!("paying at the limit failed: {error}");
988	}
989	// Last month's metered usage (scans, embeddings, storage) goes on the
990	// ledger before the month is closed and invoiced.
991	if let Err(error) = billing.charge_pending().await {
992	worker::console_error!("charging last month's metered usage failed: {error}");
993	}
994	if let Err(error) = billing.close_months().await {
995	worker::console_error!("closing the month failed: {error}");
996	}
997	if let Err(error) = billing.invoice_enterprises().await {
998	worker::console_error!("invoicing enterprises failed: {error}");
999	}
1000	if let Some(users) = &users
1001	&& let Ok(identity) = env.service("IDENTITY")
1002	&& let Err(error) = billing.warn_limits(&identity, users).await {
1003	worker::console_error!("warning owners failed: {error}");
1004	}
1005	}
1006
1007	/// Every fifteen minutes: runs settled, Stripe replayed, cards charged at
1008	/// the limit, months closed, owners warned. Once a day, at
1009	/// [`keeper::DAILY`], in an invocation of its own: the slower checks
1010	/// against Stripe and Cloudflare, and measuring.
1011	///
1012	/// Each step logs its own failure and the next still runs. The spend
1013	/// breaker and budget alerts go first: a step that runs out of CPU ends
1014	/// the invocation, and must never take the alerts with it.
1015	#[event(scheduled)]
1016	async fn scheduled(event: ScheduledEvent, env: Env, _ctx: ScheduleContext) {
1017	let Ok(billing) = Billing::from_env(&env) else {
1018	return;
1019	};
1020	// Comped budgets' alerts, and a tripped breaker staff were not told of.
1021	if let Err(error) = billing.watch_spend().await {
1022	worker::console_error!("watching g1t's own spend failed: {error}");
1023	}
1024	let keeper = keeper::Keeper::from_env(&env);
1025	// Once an hour, at the quarter past: what Cloudflare counted for the
1026	// whole platform in the hour before, against its thresholds
1027	// (platform.rs).
1028	if event.cron() == keeper::QUARTER_HOURLY && platform::hourly_due(now_ms()) {
1029	match billing.watch_platform(&keeper).await {
1030	Ok((read, breached)) => worker::console_log!("platform watch: {read} metrics, {breached} breaches"),
1031	Err(error) => worker::console_error!("watching platform usage failed: {error}"),
1032	}
1033	}
1034	let daily = event.cron() == keeper::DAILY;
1035	if !daily {
1036	tick(&billing, &env).await;
1037	}
1038	// Once a day: Stripe's endpoint kept listening to billing's events and
1039	// enabled, and saved cards and plans not read in a while read again.
1040	if daily {
1041	match billing.keep_endpoint("billing").await {
1042	Ok(done) => worker::console_log!("stripe endpoint: {done}"),
1043	Err(error) => worker::console_error!("keeping Stripe's endpoint failed: {error}"),
1044	}
1045	match billing.refresh_from_stripe().await {
1046	Ok(done) => worker::console_log!("stripe refresh: {done}"),
1047	Err(error) => worker::console_error!("refreshing from Stripe failed: {error}"),
1048	}
1049	}
1050	// Once a day, and at once if the costs were never checked: check every
1051	// cost against what Cloudflare billed.
1052	if (daily \|\| billing.never_checked().await.unwrap_or(false))
1053	&& let Err(error) = billing.reconcile(&keeper).await {
1054	worker::console_error!("checking costs against Cloudflare failed: {error}");
1055	}
1056	// Once a day: credit from g1t past its expiry stops counting
1057	// (grants.rs), before the day is reconciled.
1058	if daily {
1059	match billing.expire_credits().await {
1060	Ok(closed) => worker::console_log!("credits expired: {closed}"),
1061	Err(error) => worker::console_error!("expiring credits failed: {error}"),
1062	}
1063	}
1064	// Once a day: what Cloudflare charged, reconciled against what g1t
1065	// counted and charged; prices whose day has come; margin alerts
1066	// (margin.rs). After the keeper, so its proposals are in.
1067	if daily {
1068	match billing.costs_daily(&env, &keeper).await {
1069	Ok(run) => worker::console_log!("costs: {} lines, {} days, {} proposals, {} alerts", run.lines, run.days, run.proposals, run.alerts),
1070	Err(error) => worker::console_error!("reconciling costs failed: {error}"),
1071	}
1072	}
1073	// Once a day: what each workspace's private repositories hold, its git
1074	// operations, Deployments plans from before the g1t plan set to end,
1075	// and old reservations cleared.
1076	if daily {
1077	if let Err(error) = billing.measure_packages().await {
1078	worker::console_error!("measuring package storage failed: {error}");
1079	}
1080	if let Err(error) = billing.measure_storage().await {
1081	worker::console_error!("measuring storage failed: {error}");
1082	}
1083	if let Err(error) = billing.measure_git().await {
1084	worker::console_error!("measuring git operations failed: {error}");
1085	}
1086	if let Err(error) = billing.retire_deployments_plans().await {
1087	worker::console_error!("ending Deployments plans failed: {error}");
1088	}
1089	if let Err(error) = billing.retire_security_activations().await {
1090	worker::console_error!("ending Security and quality activations failed: {error}");
1091	}
1092	if let Err(error) = billing.sweep_reservations().await {
1093	worker::console_error!("clearing reservations failed: {error}");
1094	}
1095	// The AI Gateway's request log, from when there was one, still
1096	// keeps only its last 30 days until it is empty.
1097	let cutoff = rfc3339(now_ms().saturating_sub(30 * 24 * 60 * 60 * 1000));
1098	let cleared = match billing.db.prepare("DELETE FROM gateway_requests WHERE created_at < ?").bind(&[cutoff.into()]) {
1099	Ok(statement) => statement.run().await.map(\|_\| ()),
1100	Err(error) => Err(error),
1101	};
1102	if let Err(error) = cleared {
1103	worker::console_error!("clearing old AI Gateway requests failed: {error}");
1104	}
1105	}
1106	}
1107
1108	/// Events from the bus, on billing's own queue: only a workspace's rename
1109	/// matters here (see `rename`).
1110	#[event(queue)]
1111	async fn queue(batch: MessageBatch<g1t_contracts::events::Event>, env: Env, _ctx: Context) -> Result<()> {
1112	let billing = Billing::from_env(&env)?;
1113	let identity = env.service("IDENTITY").ok();
1114	for message in batch.messages()? {
1115	// A repository transferred: its share of the open-source pool this
1116	// month follows it. What it was charged stays with the workspace
1117	// it was charged to; usage from now on is charged to the new one.
1118	if g1t_kit::transfer::on_event(&env, &billing.db, message.body(), closing::TRANSFERRED).await? {
1119	message.ack();
1120	continue;
1121	}
1122	billing.on_event(identity.as_ref(), message.body()).await?;
1123	message.ack();
1124	}
1125	Ok(())
1126	}
1127
1128	#[event(fetch)]
1129	async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
1130	let Some(method) = rpc_method(&request) else {
1131	return Response::error("Not found", 404);
1132	};
1133	// A replica near the caller when it asks for one (crates/kit/src/d1.rs);
1134	// what other services call (can_start, start_run) asks for none.
1135	let (db, served) = g1t_kit::d1::open(&env, "DB", &request)?;
1136	let body: serde_json::Value = request.json().await?;
1137	let mut billing = Billing::from_env(&env)?;
1138	billing.db = db;
1139	let answered = match method.as_str() {
1140	"status" => reply(&billing.status()),
1141	"account" => reply(&billing.account(args(body)?).await?),
1142	"ledger" => reply(&billing.ledger(args(body)?).await?),
1143	"credits" => reply(&billing.credits(args(body)?).await?),
1144	"usage" => reply(&billing.usage(args(body)?).await?),
1145	"usage_report" => reply(&billing.usage_report(args(body)?).await?),
1146	"set_budget" => reply(&billing.set_budget(args(body)?).await?),
1147	"billing_details" => reply(&billing.billing_details(args(body)?).await?),
1148	"set_billing_details" => reply(&billing.set_billing_details(args(body)?).await?),
1149	"checkout" => reply(&billing.checkout(args(body)?).await?),
1150	"confirm" => reply(&billing.confirm(args(body)?).await?),
1151	"trial" => reply(&billing.trial(args(body)?).await?),
1152	"features" => reply(&billing.features(args(body)?).await?),
1153	"subscribe" => reply(&billing.subscribe(args(body)?).await?),
1154	"confirm_subscription" => reply(&billing.confirm_subscription(args(body)?).await?),
1155	"cancel_subscription" => reply(&billing.cancel_subscription(args(body)?).await?),
1156	"close_workspace" => reply(&billing.close_workspace(args(body)?).await?),
1157	"has_feature" => reply(&billing.has_feature(args(body)?).await?),
1158	"free_workspaces" => reply(&billing.free_workspaces(args(body)?).await?),
1159	"charge_feature" => reply(&billing.charge_feature(args(body)?).await?),
1160	"record_sandbox" => reply(&billing.record_sandbox(args(body)?).await?),
1161	"limit" => reply(&billing.limit(args(body)?).await?),
1162	"check_limit" => reply(&billing.check_limit(args(body)?).await?),
1163	"set_spend_limit" => reply(&billing.set_spend_limit(args(body)?).await?),
1164	"prices" => reply(&billing.prices().await?),
1165	"billing_portal" => reply(&billing.billing_portal(args(body)?).await?),
1166	"admin_billing_link" => reply(&billing.admin_billing_link(args(body)?).await?),
1167	"admin_stripe" => reply(&billing.admin_stripe(args(body)?).await?),
1168	"admin_enterprise_billing" => reply(&billing.admin_enterprise_billing(args(body)?).await?),
1169	"admin_enterprise_address" => reply(&billing.admin_enterprise_address(args(body)?).await?),
1170	"admin_invoice_enterprise" => reply(&billing.admin_invoice_enterprise(args(body)?).await?),
1171	"stripe_webhook" => reply(&billing.stripe_webhook(args(body)?).await?),
1172	"invoices" => reply(&billing.invoices(args(body)?).await?),
1173	"statement" => reply(&billing.statement(args(body)?).await?),
1174	"statement_entries" => reply(&billing.statement_entries(args(body)?).await?),
1175	"usage_meters" => reply(&billing.usage_meters(args(body)?).await?),
1176	"admin_workspace_invoices" => {
1177	let a: AdminWorkspaceInvoicesArgs = args(body)?;
1178	reply(&billing.workspace_invoices(&a.workspace.to_lowercase()).await?)
1179	}
1180	"admin_signals" => reply(&billing.admin_signals(args(body)?).await?),
1181	"admin_overview" => reply(&billing.admin_overview(args(body)?).await?),
1182	"admin_sales" => reply(&billing.admin_sales(args(body)?).await?),
1183	"admin_set_sales" => reply(&billing.admin_set_sales(args(body)?).await?),
1184	"admin_add_note" => reply(&billing.admin_add_note(args(body)?).await?),
1185	"admin_invoices" => reply(&billing.admin_invoices(args(body)?).await?),
1186	"admin_audit" => reply(&billing.admin_audit(args(body)?).await?),
1187	// A staff change made in another service, for sudo's audit log:
1188	// identity's restores and purges of deleted workspaces.
1189	"admin_log" => {
1190	let a: AdminLogArgs = args(body)?;
1191	billing
1192	.audit(&accounts::own_account(&a.workspace), &a.action, &a.detail, &a.by)
1193	.await?;
1194	reply(&true)
1195	}
1196	"note_pending" => reply(&billing.note_pending(args(body)?).await?),
1197	"admin_accounts" => reply(&billing.admin_accounts(args(body)?).await?),
1198	"admin_account" => reply(&billing.admin_account(args(body)?).await?),
1199	"admin_set_terms" => reply(&billing.admin_set_terms(args(body)?).await?),
1200	"admin_create_enterprise" => reply(&billing.admin_create_enterprise(args(body)?).await?),
1201	"admin_attach" => reply(&billing.admin_attach(args(body)?).await?),
1202	"admin_credit" => reply(&billing.admin_credit(args(body)?).await?),
1203	"admin_credits" => reply(&billing.admin_credits(args(body)?).await?),
1204	"admin_revoke_credit" => reply(&billing.admin_revoke_credit(args(body)?).await?),
1205	"admin_reset_billing" => reply(&billing.admin_reset_billing(&env, args(body)?).await?),
1206	"admin_set_allowances" => reply(&billing.admin_set_allowances(args(body)?).await?),
1207	"entitlements" => reply(&billing.entitlements(args(body)?).await?),
1208	"audit_retention" => reply(&billing.audit_retention(args(body)?).await?),
1209	"reserve" => reply(&billing.reserve(args(body)?).await?),
1210	"settle" => reply(&billing.settle_reservation(args(body)?).await?),
1211	"card_check" => reply(&billing.card_check(args(body)?).await?),
1212	"confirm_card_check" => reply(&billing.confirm_card_check(args(body)?).await?),
1213	"request_limit" => reply(&billing.request_limit(args(body)?).await?),
1214	"limit_requests" => reply(&billing.limit_requests(args(body)?).await?),
1215	"confirm_spike" => reply(&billing.confirm_spike(args(body)?).await?),
1216	"set_caps" => reply(&billing.set_caps(args(body)?).await?),
1217	"admin_limit_requests" => reply(&billing.admin_limit_requests(args(body)?).await?),
1218	"admin_decide_limit_request" => reply(&billing.admin_decide_limit_request(args(body)?).await?),
1219	"admin_overages" => reply(&billing.admin_overages(args(body)?).await?),
1220	"admin_goodwill" => reply(&billing.admin_goodwill(args(body)?).await?),
1221	"admin_velocity" => reply(&billing.admin_velocity(args(body)?).await?),
1222	"admin_record_payment" => reply(&billing.admin_record_payment(args(body)?).await?),
1223	"admin_costs" => reply(&billing.admin_costs(args(body)?, keeper::Keeper::from_env(&env).can_read_bill()).await?),
1224	"admin_cost_alerts" => reply(&billing.admin_cost_alerts(args(body)?).await?),
1225	"admin_spend_caps" => reply(&billing.spend_caps().await?),
1226	"admin_lift_breaker" => reply(&billing.admin_lift_breaker(args(body)?).await?),
1227	// Platform pauses (src/platform.rs): read by every service that
1228	// honours one, kept 30 seconds in each isolate.
1229	"platform_pause" => reply(&billing.pause_now().await),
1230	"admin_platform_guard" => reply(&billing.platform_guard(&keeper::Keeper::from_env(&env)).await?),
1231	"admin_set_pause" => reply(&billing.admin_set_pause(args(body)?, &keeper::Keeper::from_env(&env)).await?),
1232	"admin_decide_proposal" => reply(&billing.admin_decide_proposal(args(body)?).await?),
1233	"admin_set_cost_settings" => reply(&billing.admin_set_cost_settings(args(body)?).await?),
1234	"admin_set_cost_mapping" => reply(&billing.admin_set_cost_mapping(args(body)?).await?),
1235	"admin_run_costs" => reply(&billing.admin_run_costs(&env, args(body)?).await?),
1236	_ => Response::error("Unknown method", 404),
1237	};
1238	served.finish(answered)
1239	}
1240
1241	#[cfg(test)]
1242	mod tests {
1243	use super::*;
1244
1245	#[test]
1246	fn sandbox_time_is_charged_from_the_first_second() {
1247	// 21 millionths a second at cost, plus 20%.
1248	let price = Price::price_for(21.0, 20);
1249	assert_eq!(sandbox_charge(1, price), 26);
1250	assert_eq!(sandbox_charge(60, price), 1_512);
1251	assert_eq!(sandbox_charge(0, price), 0);
1252	}
1253
1254	#[test]
1255	fn prepaying_starts_at_twenty_five_dollars_and_bank_transfers_at_a_thousand() {
1256	assert!(prepay_amount(2_500, false).is_ok());
1257	assert!(prepay_amount(2_499, false).is_err());
1258	assert!(prepay_amount(10_000, false).is_ok());
1259	assert!(prepay_amount(1_000_000, false).is_ok());
1260	assert_eq!(prepay_amount(1_000_001, false).unwrap_err(), "Prepay between $25 and $10,000 by card; from $1,000, a bank transfer works too.");
1261	assert!(prepay_amount(99_999, true).is_err());
1262	assert!(prepay_amount(100_000, true).is_ok());
1263	}
1264
1265	#[test]
1266	fn durations_read_plainly() {
1267	assert_eq!(duration(40), "40s");
1268	assert_eq!(duration(192), "3m 12s");
1269	assert_eq!(duration(3720), "1h 2m");
1270	}
1271
1272	/// Every migration, in order, as D1 applies them.
1273	const MIGRATIONS: &[&str] = &[
1274	include_str!("../migrations/0001_init.sql"),
1275	include_str!("../migrations/0002_own_provider.sql"),
1276	include_str!("../migrations/0003_subscriptions.sql"),
1277	include_str!("../migrations/0004_sandbox_time.sql"),
1278	include_str!("../migrations/0005_limits.sql"),
1279	include_str!("../migrations/0006_prices.sql"),
1280	include_str!("../migrations/0007_pending_usage.sql"),
1281	include_str!("../migrations/0008_accounts.sql"),
1282	include_str!("../migrations/0009_autopay.sql"),
1283	include_str!("../migrations/0010_month_close.sql"),
1284	include_str!("../migrations/0011_limit_warnings.sql"),
1285	include_str!("../migrations/0012_stripe_webhooks.sql"),
1286	include_str!("../migrations/0013_invoices_trust_sales.sql"),
1287	include_str!("../migrations/0014_custom_domain_price.sql"),
1288	include_str!("../migrations/0015_sandbox_at_cost.sql"),
1289	include_str!("../migrations/0016_plans_and_pools.sql"),
1290	include_str!("../migrations/0017_flagon_comped.sql"),
1291	include_str!("../migrations/0018_one_plan.sql"),
1292	include_str!("../migrations/0019_closed_workspaces.sql"),
1293	include_str!("../migrations/0020_syntaqx_standard.sql"),
1294	include_str!("../migrations/0021_no_quotas.sql"),
1295	include_str!("../migrations/0022_costs_and_margin.sql"),
1296	include_str!("../migrations/0023_one_operation_mapping.sql"),
1297	include_str!("../migrations/0024_spend_caps.sql"),
1298	include_str!("../migrations/0025_run_tier.sql"),
1299	include_str!("../migrations/0026_ledger_by_workspace_time.sql"),
1300	include_str!("../migrations/0027_stripe_cache.sql"),
1301	include_str!("../migrations/0028_drop_stripe_webhooks.sql"),
1302	include_str!("../migrations/0029_audit_retention.sql"),
1303	include_str!("../migrations/0030_token_usage.sql"),
1304	include_str!("../migrations/0031_package_storage.sql"),
1305	include_str!("../migrations/0032_workspace_value.sql"),
1306	include_str!("../migrations/0033_given_away.sql"),
1307	include_str!("../migrations/0034_given_by_why.sql"),
1308	include_str!("../migrations/0035_cf_subscriptions.sql"),
1309	include_str!("../migrations/0036_model_costs_in_full.sql"),
1310	include_str!("../migrations/0037_security_activation.sql"),
1311	include_str!("../migrations/0038_staff_credits.sql"),
1312	include_str!("../migrations/0039_discounts_not_comped.sql"),
1313	include_str!("../migrations/0040_ai_credit.sql"),
1314	include_str!("../migrations/0041_agent_rate_own_key.sql"),
1315	include_str!("../migrations/0042_agent_rate_weights.sql"),
1316	include_str!("../migrations/0043_tax_and_card_fees.sql"),
1317	include_str!("../migrations/0044_cache_reads_count_a_tenth.sql"),
1318	include_str!("../migrations/0045_gateway.sql"),
1319	include_str!("../migrations/0046_reset_costs.sql"),
1320	include_str!("../migrations/0047_gateway_formats.sql"),
1321	include_str!("../migrations/0048_model_catalogue.sql"),
1322	include_str!("../migrations/0049_superseded_proposals.sql"),
1323	include_str!("../migrations/0050_ledger_usage_by_time.sql"),
1324	include_str!("../migrations/0051_platform_guardrails.sql"),
1325	include_str!("../migrations/0052_cloudflare_cycle.sql"),
1326	include_str!("../migrations/0053_models_at_cost.sql"),
1327	include_str!("../migrations/0054_agent_attribution.sql"),
1328	include_str!("../migrations/0055_app_containers.sql"),
1329	];
1330
1331	/// The columns of `table` after the migrations: each with whether an
1332	/// insert must give it (NOT NULL, no default).
1333	fn columns(table: &str) -> Vec<(String, bool)> {
1334	let mut out = vec![];
1335	for sql in MIGRATIONS {
1336	let sql: String = sql.lines().map(\|l\| l.split("--").next().unwrap_or("")).collect::<Vec<_>>().join("\n");
1337	if let Some(start) = sql.find(&format!("CREATE TABLE {table} (")) {
1338	let body = &sql[start + format!("CREATE TABLE {table} (").len()..];
1339	let body = &body[..body.find(");").unwrap()];
1340	for column in body.split(",\n") {
1341	let column = column.trim();
1342	let name = column.split_whitespace().next().unwrap_or("").to_owned();
1343	if name.is_empty() \|\| name == "PRIMARY" {
1344	continue;
1345	}
1346	let required = column.contains("NOT NULL") && !column.contains("DEFAULT") && !column.contains("PRIMARY KEY");
1347	out.push((name, required \|\| column.contains("PRIMARY KEY")));
1348	}
1349	}
1350	for statement in sql.split(';') {
1351	let s = statement.trim();
1352	if let Some(rest) = s.strip_prefix(&format!("ALTER TABLE {table} ADD COLUMN ")) {
1353	let name = rest.split_whitespace().next().unwrap().to_owned();
1354	out.push((name, rest.contains("NOT NULL") && !rest.contains("DEFAULT")));
1355	}
1356	}
1357	}
1358	out
1359	}
1360
1361	/// The columns an `INSERT INTO table (…)` names.
1362	fn inserted(sql: &str, table: &str) -> Vec<String> {
1363	let start = sql.find(&format!("INSERT INTO {table} (")).unwrap() + format!("INSERT INTO {table} (").len();
1364	sql[start..start + sql[start..].find(')').unwrap()].split(',').map(\|c\| c.trim().to_owned()).collect()
1365	}
1366
1367
1368	#[test]
1369	fn every_checkout_insert_fills_the_table() {
1370	// The plan's, the activation's, a card check's and a prepayment's
1371	// pages all go through one insert, and it names every
1372	// column the table needs and none it lacks: a missing NOT NULL
1373	// column, or one from a migration not yet applied, was a 500.
1374	let table = columns("checkouts");
1375	let names: Vec<&str> = table.iter().map(\|(n, _)\| n.as_str()).collect();
1376	assert_eq!(names, ["id", "workspace", "amount_cents", "created_by", "status", "created_at", "feature", "fee_cents"]);
1377	let given = inserted(CHECKOUT_INSERT, "checkouts");
1378	for column in &given {
1379	assert!(names.contains(&column.as_str()), "the insert names {column}, which checkouts does not have");
1380	}
1381	for (column, required) in &table {
1382	assert!(!required \|\| given.contains(column), "checkouts needs {column}, which the insert leaves out");
1383	}
1384	// As many values as columns.
1385	let values = CHECKOUT_INSERT.split("VALUES").nth(1).unwrap();
1386	assert_eq!(values.matches('?').count() + values.matches("'open'").count(), given.len());
1387	// And no page writes a row of its own any more.
1388	for (file, source) in [
1389	("lib.rs", include_str!("lib.rs")),
1390	("features.rs", include_str!("features.rs")),
1391	("cards.rs", include_str!("cards.rs")),
1392	] {
1393	let inserts = source.matches(concat!("INSERT INTO ", "checkouts")).count();
1394	assert!(inserts <= usize::from(file == "lib.rs"), "{file} inserts into checkouts on its own");
1395	}
1396	}
1397
1398	#[test]
1399	fn a_new_ledger_column_has_a_default() {
1400	// The ledger is written from a dozen places; a column without a
1401	// default would break every one that does not name it.
1402	for (column, required) in columns("ledger") {
1403	assert!(!required \|\| ["id", "workspace", "kind", "amount_micros", "description", "reference", "created_at"].contains(&column.as_str()), "{column}");
1404	}
1405	}
1406	}