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