flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

git agents cloudflare rust forge mcp

1,574 linesCode BlameRaw

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