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