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.

g1t.sh

git agents cloudflare rust forge mcp

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