Skip to content
Sign in

flagon-io/g1t

public
0
927 linesCodeBlameRaw
1//! Stripe telling billing what happened, and enterprise invoices.
2//!
3//! Most of billing asks Stripe when it needs to know: a payment page is
4//! confirmed when the person comes back, a plan is checked when its period
5//! ends. That misses whatever happens while no one is looking: a page paid
6//! for and closed, a renewal that failed, a refund, a dispute, an invoice
7//! paid by bank transfer a week later. Stripe sends each as an event to
8//! `https://api.g1t.sh/stripe/webhook`; the API passes the raw body and its
9//! signature here, untouched.
10//!
11//! - The endpoint is registered by billing itself, from sudo, once per
12//! mode, and its signing secret is kept in billing's database. It is never
13//! shown, and nothing can be posted here without it.
14//! - Each event is handled once, by id, and recorded with what was done.
15//! - Every handler is safe alongside the paths that ask Stripe directly:
16//! both claim the same rows.
17//!
18//! Enterprises are invoiced: one Stripe invoice per month (or sooner, from
19//! sudo), with a line per workspace for what it owes, sent to the
20//! enterprise's billing email and paid on Stripe's hosted invoice page.
21//! When it is paid, each workspace is credited its line; when it goes
22//! overdue, their work stops until it is paid.
23
24use g1t_contracts::billing::{
25 AdminEnterpriseBillingArgs, AdminInvoiceEnterpriseArgs, AdminStripeArgs, BillingAccount, EnterpriseInvoice,
26 EntryKind, InvoiceLine, StripeEventSummary, StripeStatus, StripeWebhook, StripeWebhookArgs,
27};
28use g1t_contracts::time::rfc3339;
29use g1t_contracts::{FailureCode, Outcome};
30use g1t_kit::now_ms;
31use hmac::{Hmac, Mac};
32use serde::Deserialize;
33use serde_json::Value;
34use sha2::Sha256;
35use worker::Result;
36use worker::wasm_bindgen::JsValue;
37
38use crate::Billing;
39
40/// Where Stripe sends events.
41pub(crate) const WEBHOOK_URL: &str = "https://api.g1t.sh/stripe/webhook";
42
43/// The events billing acts on.
44pub(crate) const EVENTS: &[&str] = &[
45 "checkout.session.completed",
46 // A prepayment by bank transfer: the page completes when the transfer
47 // is set up, and this comes when the money arrives.
48 "checkout.session.async_payment_succeeded",
49 "customer.subscription.updated",
50 "customer.subscription.deleted",
51 "invoice.paid",
52 "invoice.payment_failed",
53 "invoice.overdue",
54 "invoice.voided",
55 "charge.refunded",
56 "charge.dispute.created",
57 "charge.dispute.closed",
58 // The saved card, kept on the account row (stripe_sync.rs).
59 "customer.updated",
60 "payment_method.attached",
61 "payment_method.detached",
62 "payment_method.updated",
63 "payment_method.automatically_updated",
64 "setup_intent.succeeded",
65];
66
67/// Events billing handles that `has` does not include, in billing's order.
68pub(crate) fn missing_events(has: &[String]) -> Vec<String> {
69 // `*` is every event.
70 if has.iter().any(|event| event == "*") {
71 return Vec::new();
72 }
73 EVENTS.iter().filter(|event| !has.iter().any(|h| h == *event)).map(|event| (*event).to_owned()).collect()
74}
75
76/// How old a signed event may be, so a captured one cannot be replayed.
77const TOLERANCE_SECONDS: i64 = 5 * 60;
78
79/// Whether `header` (`t=…,v1=…`) signs `payload` with `secret`, within the
80/// tolerance of `now_seconds`.
81pub(crate) fn verify(payload: &str, header: &str, secret: &str, now_seconds: i64) -> bool {
82 let mut timestamp = None;
83 let mut signatures = vec![];
84 for part in header.split(',') {
85 match part.trim().split_once('=') {
86 Some(("t", value)) => timestamp = value.parse::<i64>().ok(),
87 Some(("v1", value)) => signatures.push(value.to_owned()),
88 _ => {}
89 }
90 }
91 let Some(timestamp) = timestamp else { return false };
92 if (now_seconds - timestamp).abs() > TOLERANCE_SECONDS {
93 return false;
94 }
95 let Ok(mut mac) = Hmac::<Sha256>::new_from_slice(secret.as_bytes()) else { return false };
96 mac.update(format!("{timestamp}.{payload}").as_bytes());
97 let expected = mac.finalize().into_bytes();
98 signatures.iter().any(|signature| {
99 hex::decode(signature).is_ok_and(|given| {
100 // Constant time: compare every byte whatever the first difference.
101 given.len() == expected.len() && given.iter().zip(expected.iter()).fold(0u8, |acc, (a, b)| acc | (a ^ b)) == 0
102 })
103 })
104}
105
106/// The destination at billing's address, as Stripe lists it.
107#[derive(Deserialize)]
108pub(crate) struct Destination {
109 pub(crate) id: String,
110 url: String,
111 /// `enabled` or `disabled`.
112 pub(crate) status: String,
113 pub(crate) enabled_events: Vec<String>,
114 created: i64,
115}
116
117#[derive(Deserialize)]
118struct EventRow {
119 id: String,
120 r#type: String,
121 outcome: String,
122 received_at: String,
123}
124
125#[derive(Deserialize)]
126struct InvoiceRow {
127 invoice_id: String,
128 period: String,
129 amount_micros: i64,
130 status: String,
131 hosted_url: Option<String>,
132 created_at: String,
133}
134
135#[derive(Deserialize)]
136struct LineRow {
137 workspace: String,
138 amount_micros: i64,
139}
140
141/// The idempotency key of an enterprise's invoice: the same enterprise,
142/// period and amounts are the same invoice however often it is attempted.
143pub(crate) fn enterprise_invoice_key(id: &str, period: &str, lines: &[(&str, i64)]) -> String {
144 // FNV-1a over every line, so the key stays short however many there are.
145 let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
146 for (workspace, cents) in lines {
147 for byte in format!("{workspace}={cents};").bytes() {
148 hash = (hash ^ u64::from(byte)).wrapping_mul(0x0100_0000_01b3);
149 }
150 }
151 format!("ent-invoice/{id}/{period}/{hash:016x}")
152}
153
154impl Billing {
155 pub(crate) fn mode(&self) -> &'static str {
156 match &self.stripe {
157 None => "off",
158 Some(stripe) if stripe.live() => "live",
159 Some(_) => "test",
160 }
161 }
162
163 /// The destination at billing's address in Stripe, for this key's
164 /// mode: an enabled one first. None when there is none.
165 pub(crate) async fn destination(&self) -> Result<Option<Destination>> {
166 let Some(stripe) = &self.stripe else { return Ok(None) };
167 #[derive(Deserialize)]
168 struct List {
169 data: Vec<Destination>,
170 }
171 let mut ours: Vec<Destination> =
172 stripe.get::<List>("/webhook_endpoints?limit=100").await?.data.into_iter().filter(|d| d.url == WEBHOOK_URL).collect();
173 ours.sort_by_key(|d| d.status != "enabled");
174 Ok(ours.into_iter().next())
175 }
176
177 // --- Staff ------------------------------------------------------------
178
179 pub(crate) async fn admin_stripe(&self, a: AdminStripeArgs) -> Result<StripeStatus> {
180 let mut error = None;
181 if a.fix
182 && let Err(e) = self.keep_endpoint(a.by.as_deref().unwrap_or("sudo")).await
183 {
184 error = Some(e.to_string());
185 }
186 let (webhook, missing_events) = match self.destination().await {
187 Ok(Some(d)) => {
188 let missing = missing_events(&d.enabled_events);
189 let webhook = StripeWebhook {
190 url: d.url,
191 endpoint_id: d.id,
192 status: d.status,
193 events: d.enabled_events,
194 created_at: rfc3339(d.created.max(0) as u64 * 1000),
195 };
196 (Some(webhook), missing)
197 }
198 Ok(None) => (None, Vec::new()),
199 Err(e) => {
200 error = error.or(Some(format!("Stripe could not be read: {e}")));
201 (None, Vec::new())
202 }
203 };
204 let recent_events = self
205 .db
206 .prepare("SELECT * FROM stripe_events ORDER BY received_at DESC LIMIT 25")
207 .all()
208 .await?
209 .results::<EventRow>()?
210 .into_iter()
211 .map(|row| StripeEventSummary { id: row.id, kind: row.r#type, outcome: row.outcome, received_at: row.received_at })
212 .collect();
213 Ok(StripeStatus {
214 mode: self.mode().to_owned(),
215 secret_set: self.webhook_secret.is_some(),
216 webhook,
217 missing_events,
218 recent_events,
219 error,
220 })
221 }
222
223 // --- Events -----------------------------------------------------------
224
225 pub(crate) async fn stripe_webhook(&self, a: StripeWebhookArgs) -> Result<Outcome<bool>> {
226 let Some(secret) = &self.webhook_secret else {
227 return Ok(Outcome::fail(
228 FailureCode::Conflict,
229 "STRIPE_WEBHOOK_SECRET is not set, so billing cannot check that events come from Stripe.",
230 ));
231 };
232 let now_seconds = (now_ms() / 1000) as i64;
233 if !verify(&a.payload, &a.signature, secret, now_seconds) {
234 return Ok(Outcome::fail(FailureCode::Forbidden, "The signature does not match."));
235 }
236 let event: Value = serde_json::from_str(&a.payload).map_err(|e| worker::Error::RustError(e.to_string()))?;
237 if event["id"].as_str().unwrap_or_default().is_empty() {
238 return Ok(Outcome::fail(FailureCode::Invalid, "Not an event."));
239 }
240 Ok(Outcome::Ok(self.process_event(&event).await?))
241 }
242
243 /// Handles a Stripe event once, however often it arrives: by webhook,
244 /// or again from the event list (stripe_sync.rs). False when it was
245 /// seen before.
246 pub(crate) async fn process_event(&self, event: &Value) -> Result<bool> {
247 let id = event["id"].as_str().unwrap_or_default().to_owned();
248 let kind = event["type"].as_str().unwrap_or_default().to_owned();
249 // Once each: the first to record it handles it.
250 let claimed = self
251 .db
252 .prepare("INSERT OR IGNORE INTO stripe_events (id, type, outcome, received_at) VALUES (?, ?, 'handling', ?) RETURNING id")
253 .bind(&[id.as_str().into(), kind.as_str().into(), rfc3339(now_ms()).into()])?
254 .first::<Value>(None)
255 .await?;
256 if claimed.is_none() {
257 return Ok(false);
258 }
259 let outcome = match self.handle(&kind, event).await {
260 Ok(outcome) => outcome,
261 Err(error) => {
262 // Let Stripe send it again: forget it was seen.
263 self.db.prepare("DELETE FROM stripe_events WHERE id = ?").bind(&[id.as_str().into()])?.run().await?;
264 return Err(error);
265 }
266 };
267 self.db
268 .prepare("UPDATE stripe_events SET outcome = ? WHERE id = ?")
269 .bind(&[outcome.as_str().into(), id.as_str().into()])?
270 .run()
271 .await?;
272 Ok(true)
273 }
274
275 async fn handle(&self, kind: &str, event: &Value) -> Result<String> {
276 let object = &event["data"]["object"];
277 let text = |key: &str| object[key].as_str().unwrap_or_default().to_owned();
278 // The customer whose saved card may have changed. A detached card
279 // has no customer any more; the event says whose it was.
280 let card_owner = match kind {
281 "customer.updated" => object["id"].as_str(),
282 "payment_method.detached" => event["data"]["previous_attributes"]["customer"].as_str(),
283 _ => object["customer"].as_str(),
284 };
285 Ok(match kind {
286 "checkout.session.completed" | "checkout.session.async_payment_succeeded" => self.settle_checkout(&text("id")).await?,
287 "customer.subscription.updated" | "customer.subscription.deleted" => {
288 self.settle_subscription(&text("id")).await?
289 }
290 "invoice.paid" => {
291 if let Some(subscription) = object["subscription"].as_str() {
292 self.plan_paid(subscription, &text("id"), object["amount_paid"].as_i64().unwrap_or(0)).await?;
293 self.settle_subscription(subscription).await?
294 } else if let Some(done) = self.workspace_invoice_paid(&text("id")).await? {
295 done
296 } else {
297 self.enterprise_invoice_paid(&text("id")).await?
298 }
299 }
300 "invoice.payment_failed" => match (object["subscription"].as_str(), object["metadata"]["g1t_workspace"].as_str()) {
301 (Some(subscription), _) => self.settle_subscription(subscription).await?,
302 (None, Some(tagged)) => {
303 // The invoice's own row names the workspace as it is
304 // now; the metadata keeps the slug it was sent under.
305 let workspace = self.workspace_of_invoice(&text("id")).await?.unwrap_or_else(|| tagged.to_owned());
306 let workspace = workspace.as_str();
307 self.mark_declined(workspace, "the card was declined for an invoice").await?;
308 format!("{workspace}: invoice payment failed; work stopped")
309 }
310 _ => "ignored: not g1t's".to_owned(),
311 },
312 "invoice.overdue" => self.enterprise_invoice_status(&text("id"), "overdue").await?,
313 "invoice.voided" => self.enterprise_invoice_status(&text("id"), "void").await?,
314 "charge.refunded" => self.refunded(object).await?,
315 "charge.dispute.created" => self.disputed(object, true).await?,
316 "charge.dispute.closed" => self.disputed(object, object["status"].as_str() == Some("lost")).await?,
317 "customer.updated"
318 | "payment_method.attached"
319 | "payment_method.detached"
320 | "payment_method.updated"
321 | "payment_method.automatically_updated"
322 | "setup_intent.succeeded" => match card_owner {
323 Some(customer) => self.sync_card_of(customer).await?,
324 None => "ignored: no customer".to_owned(),
325 },
326 _ => "ignored".to_owned(),
327 })
328 }
329
330 /// A payment page done, whether or not the person came back to g1t.
331 async fn settle_checkout(&self, session_id: &str) -> Result<String> {
332 #[derive(Deserialize)]
333 struct Open {
334 workspace: String,
335 created_by: String,
336 feature: Option<String>,
337 }
338 let Some(open) = self
339 .db
340 .prepare("SELECT workspace, created_by, feature FROM checkouts WHERE id = ? AND status = 'open'")
341 .bind(&[session_id.into()])?
342 .first::<Open>(None)
343 .await?
344 else {
345 return Ok("ignored: already settled or not g1t's".to_owned());
346 };
347 // A card check: saved and verified, never charged.
348 if open.feature.as_deref() == Some(crate::cards::CARD_CHECK) {
349 return Ok(match self.settle_card_check(session_id).await? {
350 Ok(done) => done,
351 Err(why) => format!("card check not passed: {why}"),
352 });
353 }
354 let Some(stripe) = &self.stripe else { return Ok("ignored: payments off".to_owned()) };
355 let session = stripe.session(session_id).await?;
356 if session.payment_status != "paid" && open.feature.is_none() {
357 return Ok("ignored: not paid".to_owned());
358 }
359 let claimed = self
360 .db
361 .prepare("UPDATE checkouts SET status = 'paid' WHERE id = ? AND status = 'open' RETURNING id")
362 .bind(&[session_id.into()])?
363 .first::<Value>(None)
364 .await?;
365 if claimed.is_none() {
366 return Ok("ignored: settled meanwhile".to_owned());
367 }
368 match open.feature.as_deref() {
369 None => {
370 let cents = i64::from(session.amount_total.unwrap_or(0));
371 self.enter(
372 &open.workspace,
373 EntryKind::TopUp,
374 cents * 10_000,
375 "Paid in advance",
376 &session.id,
377 None,
378 None,
379 Some(&open.created_by),
380 session.customer.as_deref(),
381 )
382 .await?;
383 Ok(format!("credited {} to {}", crate::features::dollars(cents * 10_000), open.workspace))
384 }
385 Some(feature) => {
386 let Some(feature) = g1t_contracts::billing::Feature::parse(feature) else {
387 return Ok("ignored: unknown feature".to_owned());
388 };
389 if let Some(subscription_id) = &session.subscription {
390 let subscription = stripe.subscription(subscription_id).await?;
391 self.record(&open.workspace, feature, &subscription, &open.created_by).await?;
392 }
393 self.db
394 .prepare(
395 "INSERT INTO accounts (workspace, balance_micros, customer_id, created_at) VALUES (?1, 0, ?2, ?3)
396 ON CONFLICT (workspace) DO UPDATE SET customer_id = COALESCE(customer_id, ?2)",
397 )
398 .bind(&[open.workspace.as_str().into(), crate::optional(session.customer.as_deref()), rfc3339(now_ms()).into()])?
399 .run()
400 .await?;
401 Ok(format!("{} plan started for {}", feature.title(), open.workspace))
402 }
403 }
404 }
405
406 /// The plan's monthly price, paid: revenue that never goes through the
407 /// ledger, recorded once per invoice for sudo's figures and for trust.
408 async fn plan_paid(&self, subscription_id: &str, invoice_id: &str, amount_cents: i64) -> Result<()> {
409 if amount_cents <= 0 || invoice_id.is_empty() {
410 return Ok(());
411 }
412 self.db
413 .prepare(
414 "INSERT INTO plan_payments (invoice_id, workspace, amount_micros, paid_at)
415 SELECT ?1, workspace, ?2, ?3 FROM subscriptions WHERE subscription_id = ?4
416 ON CONFLICT (invoice_id) DO NOTHING",
417 )
418 .bind(&[
419 invoice_id.into(),
420 ((amount_cents * 10_000) as f64).into(),
421 rfc3339(now_ms()).into(),
422 subscription_id.into(),
423 ])?
424 .run()
425 .await?;
426 Ok(())
427 }
428
429 /// A plan that changed at Stripe: renewed, failed, canceled.
430 pub(crate) async fn settle_subscription(&self, subscription_id: &str) -> Result<String> {
431 #[derive(Deserialize)]
432 struct Plan {
433 workspace: String,
434 feature: String,
435 started_by: String,
436 }
437 let Some(plan) = self
438 .db
439 .prepare("SELECT workspace, feature, started_by FROM subscriptions WHERE subscription_id = ?")
440 .bind(&[subscription_id.into()])?
441 .first::<Plan>(None)
442 .await?
443 else {
444 return Ok("ignored: not a g1t plan".to_owned());
445 };
446 let (Some(stripe), Some(feature)) = (&self.stripe, g1t_contracts::billing::Feature::parse(&plan.feature)) else {
447 return Ok("ignored".to_owned());
448 };
449 let subscription = stripe.subscription(subscription_id).await?;
450 self.record(&plan.workspace, feature, &subscription, &plan.started_by).await?;
451 Ok(format!("{} plan for {} is {}", feature.title(), plan.workspace, subscription.status))
452 }
453
454 /// The workspace a Stripe customer belongs to.
455 pub(crate) async fn workspace_of_customer(&self, customer: &str) -> Result<Option<String>> {
456 #[derive(Deserialize)]
457 struct Row {
458 workspace: String,
459 }
460 Ok(self
461 .db
462 .prepare("SELECT workspace FROM accounts WHERE customer_id = ?")
463 .bind(&[customer.into()])?
464 .first::<Row>(None)
465 .await?
466 .map(|row| row.workspace))
467 }
468
469 /// The workspace a workspace invoice was sent to, under its slug now.
470 async fn workspace_of_invoice(&self, invoice_id: &str) -> Result<Option<String>> {
471 #[derive(Deserialize)]
472 struct Row {
473 workspace: String,
474 }
475 Ok(self
476 .db
477 .prepare("SELECT workspace FROM workspace_invoices WHERE invoice_id = ?")
478 .bind(&[invoice_id.into()])?
479 .first::<Row>(None)
480 .await?
481 .map(|row| row.workspace))
482 }
483
484 /// Money given back: what was paid is less, by the refund.
485 async fn refunded(&self, charge: &Value) -> Result<String> {
486 let Some(customer) = charge["customer"].as_str() else { return Ok("ignored: no customer".to_owned()) };
487 let Some(workspace) = self.workspace_of_customer(customer).await? else {
488 return Ok("ignored: not a workspace's customer".to_owned());
489 };
490 let refunded = charge["amount_refunded"].as_i64().unwrap_or(0);
491 if refunded <= 0 {
492 return Ok("ignored: nothing refunded".to_owned());
493 }
494 // Each refund total once, so partial refunds add up correctly.
495 let charge_id = charge["id"].as_str().unwrap_or_default();
496 #[derive(Deserialize)]
497 struct Sum {
498 micros: Option<i64>,
499 }
500 let already = self
501 .db
502 .prepare("SELECT -SUM(amount_micros) AS micros FROM ledger WHERE reference LIKE ?")
503 .bind(&[format!("refund/{charge_id}/%").into()])?
504 .first::<Sum>(None)
505 .await?
506 .and_then(|s| s.micros)
507 .unwrap_or(0);
508 let new = refunded * 10_000 - already;
509 if new <= 0 {
510 return Ok("ignored: refund already recorded".to_owned());
511 }
512 self.enter(
513 &workspace,
514 EntryKind::TopUp,
515 -new,
516 "Refunded to the card",
517 &format!("refund/{charge_id}/{refunded}"),
518 None,
519 None,
520 None,
521 None,
522 )
523 .await?;
524 Ok(format!("refund of {} recorded for {workspace}", crate::features::dollars(new)))
525 }
526
527 /// A disputed payment stops the workspace's work until it is resolved;
528 /// one closed in the workspace's favour lets it go on.
529 async fn disputed(&self, dispute: &Value, stop: bool) -> Result<String> {
530 let Some(stripe) = &self.stripe else { return Ok("ignored".to_owned()) };
531 let Some(charge_id) = dispute["charge"].as_str() else { return Ok("ignored: no charge".to_owned()) };
532 let charge: Value = stripe.get(&format!("/charges/{charge_id}")).await?;
533 let Some(workspace) = (match charge["customer"].as_str() {
534 Some(customer) => self.workspace_of_customer(customer).await?,
535 None => None,
536 }) else {
537 return Ok("ignored: not a workspace's customer".to_owned());
538 };
539 let now = rfc3339(now_ms());
540 // The disputed payment never counts toward trust again.
541 for reference in [charge["payment_intent"].as_str(), charge["invoice"].as_str()].into_iter().flatten() {
542 self.db
543 .prepare("UPDATE ledger SET disputed = ? WHERE workspace = ? AND reference = ?")
544 .bind(&[(if stop { 1 } else { 0 }).into(), workspace.as_str().into(), reference.into()])?
545 .run()
546 .await?;
547 }
548 if stop {
549 self.db
550 .prepare(
551 "INSERT INTO limits (workspace, autopay_failed_at, autopay_error, updated_at) VALUES (?1, ?2, ?3, ?2)
552 ON CONFLICT (workspace) DO UPDATE SET autopay_failed_at = ?2, autopay_error = ?3, updated_at = ?2",
553 )
554 .bind(&[workspace.as_str().into(), now.as_str().into(), "a payment was disputed with the card's bank".into()])?
555 .run()
556 .await?;
557 } else {
558 self.db
559 .prepare("UPDATE limits SET autopay_failed_at = NULL, autopay_error = NULL WHERE workspace = ?")
560 .bind(&[workspace.as_str().into()])?
561 .run()
562 .await?;
563 }
564 let account = self.account_of(&workspace).await?;
565 let what = if stop { "dispute: work stopped" } else { "dispute closed in the workspace's favour" };
566 self.audit(&account.id, "dispute", &format!("{workspace}: {what}"), "stripe").await?;
567 Ok(format!("{workspace}: {what}"))
568 }
569
570 // --- Enterprise invoices ------------------------------------------------
571
572 pub(crate) async fn admin_enterprise_billing(&self, a: AdminEnterpriseBillingArgs) -> Result<Outcome<BillingAccount>> {
573 let email = a.email.trim().to_lowercase();
574 if !email.contains('@') || email.contains(char::is_whitespace) || a.by.trim().is_empty() {
575 return Ok(Outcome::fail(FailureCode::Invalid, "Give the email the enterprise's invoices go to."));
576 }
577 let Some(stripe) = &self.stripe else {
578 return Ok(Outcome::fail(FailureCode::Conflict, "Payments are not set up on this g1t."));
579 };
580 #[derive(Deserialize)]
581 struct Row {
582 name: String,
583 kind: String,
584 customer_id: Option<String>,
585 }
586 let Some(row) = self
587 .db
588 .prepare("SELECT name, kind, customer_id FROM billing_accounts WHERE id = ?")
589 .bind(&[a.id.as_str().into()])?
590 .first::<Row>(None)
591 .await?
592 .filter(|row| row.kind == "enterprise")
593 else {
594 return Ok(Outcome::fail(FailureCode::NotFound, "No such enterprise."));
595 };
596 #[derive(Deserialize)]
597 struct Customer {
598 id: String,
599 }
600 let fields = [
601 ("name", row.name.clone()),
602 ("email", email.clone()),
603 ("metadata[g1t_enterprise]", a.id.clone()),
604 ];
605 let customer: Customer = match &row.customer_id {
606 Some(id) => stripe.post(&format!("/customers/{id}"), &fields).await?,
607 None => stripe.post("/customers", &fields).await?,
608 };
609 self.db
610 .prepare("UPDATE billing_accounts SET billing_email = ?, customer_id = ? WHERE id = ?")
611 .bind(&[email.as_str().into(), customer.id.as_str().into(), a.id.as_str().into()])?
612 .run()
613 .await?;
614 self.audit(&a.id, "billing_email", &format!("Invoices go to {email}"), &a.by).await?;
615 Ok(match self.enterprise(&a.id).await? {
616 Some(account) => Outcome::Ok(account),
617 None => Outcome::fail(FailureCode::NotFound, "No such enterprise."),
618 })
619 }
620
621 pub(crate) async fn admin_invoice_enterprise(&self, a: AdminInvoiceEnterpriseArgs) -> Result<Outcome<EnterpriseInvoice>> {
622 if a.by.trim().is_empty() {
623 return Ok(Outcome::fail(FailureCode::Invalid, "Say who is sending it."));
624 }
625 match self.invoice_enterprise(&a.id, "now", &a.by).await? {
626 Ok(invoice) => Ok(Outcome::Ok(invoice)),
627 Err(why) => Ok(Outcome::fail(FailureCode::Conflict, why)),
628 }
629 }
630
631 /// Invoices each enterprise for the month that closed. Live payments
632 /// only; sudo can send one sooner in test mode.
633 pub(crate) async fn invoice_enterprises(&self) -> Result<()> {
634 if !self.stripe.as_ref().is_some_and(crate::stripe::Stripe::live) {
635 return Ok(());
636 }
637 let closing = crate::limits::previous_month(&rfc3339(now_ms())[..7]);
638 #[derive(Deserialize)]
639 struct Id {
640 id: String,
641 }
642 let due = self
643 .db
644 .prepare(format!(
645 "SELECT id FROM billing_accounts WHERE kind = 'enterprise' AND customer_id IS NOT NULL
646 AND NOT {full}
647 AND NOT EXISTS (SELECT 1 FROM enterprise_invoices i WHERE i.account_id = billing_accounts.id AND i.period = ?)",
648 full = crate::sales::FULL_DISCOUNT_SQL
649 ))
650 .bind(&[closing.as_str().into()])?
651 .all()
652 .await?
653 .results::<Id>()?;
654 for Id { id } in due {
655 if let Err(why) = self.invoice_enterprise(&id, &closing, "month close").await? {
656 worker::console_log!("enterprise {id} not invoiced for {closing}: {why}");
657 }
658 }
659 Ok(())
660 }
661
662 /// One invoice for what each of the enterprise's workspaces owes now.
663 async fn invoice_enterprise(&self, id: &str, period: &str, by: &str) -> Result<std::result::Result<EnterpriseInvoice, String>> {
664 let Some(stripe) = &self.stripe else { return Ok(Err("Payments are not set up.".into())) };
665 let Some(account) = self.enterprise(id).await? else { return Ok(Err("No such enterprise.".into())) };
666 #[derive(Deserialize)]
667 struct Customer {
668 customer_id: Option<String>,
669 }
670 let Some(customer) = self
671 .db
672 .prepare("SELECT customer_id FROM billing_accounts WHERE id = ?")
673 .bind(&[id.into()])?
674 .first::<Customer>(None)
675 .await?
676 .and_then(|row| row.customer_id)
677 else {
678 return Ok(Err("Set where the enterprise's invoices go first.".into()));
679 };
680 // What each workspace owes: its charges less what it has paid, and
681 // less what is on invoices still open.
682 let mut lines = vec![];
683 for workspace in &account.workspaces {
684 let balance = self.row(workspace).await?.map_or(0, |row| row.balance_micros);
685 #[derive(Deserialize)]
686 struct Sum {
687 micros: Option<i64>,
688 }
689 let invoiced = self
690 .db
691 .prepare(
692 "SELECT SUM(l.amount_micros) AS micros FROM enterprise_invoice_lines l
693 JOIN enterprise_invoices i ON i.invoice_id = l.invoice_id
694 WHERE l.workspace = ? AND i.status IN ('open', 'overdue')",
695 )
696 .bind(&[workspace.as_str().into()])?
697 .first::<Sum>(None)
698 .await?
699 .and_then(|s| s.micros)
700 .unwrap_or(0);
701 let owed = (-balance).max(0) - invoiced;
702 if owed >= 10_000 {
703 lines.push(InvoiceLine { workspace: workspace.clone(), amount_micros: owed });
704 }
705 }
706 if lines.is_empty() {
707 return Ok(Err("Its workspaces owe nothing to invoice.".into()));
708 }
709 let fields = [
710 ("customer", customer.clone()),
711 ("collection_method", "send_invoice".to_owned()),
712 ("days_until_due", "30".to_owned()),
713 ("pending_invoice_items_behavior", "exclude".to_owned()),
714 ("description", format!("g1t usage for the {} enterprise", account.name)),
715 ("metadata[g1t_enterprise]", id.to_owned()),
716 ("metadata[period]", period.to_owned()),
717 ];
718 #[derive(Deserialize)]
719 struct Invoice {
720 id: String,
721 #[serde(default)]
722 hosted_invoice_url: Option<String>,
723 #[serde(default)]
724 amount_due: i64,
725 }
726 // The draft first, keyed on what it bills, and its lines put on it:
727 // an attempt that failed half way is found again, never billed again
728 // by an invoice sweeping its pending lines in.
729 let cents: Vec<i64> = lines.iter().map(|line| (line.amount_micros + 9_999) / 10_000).collect();
730 let keyed: Vec<(&str, i64)> = lines.iter().zip(&cents).map(|(line, cents)| (line.workspace.as_str(), *cents)).collect();
731 let key = enterprise_invoice_key(id, period, &keyed);
732 let draft: Invoice = stripe.post_idempotent("/invoices", &fields, &key).await?;
733 for (line, cents) in lines.iter().zip(&cents) {
734 let fields = [
735 ("customer", customer.clone()),
736 ("invoice", draft.id.clone()),
737 ("amount", cents.to_string()),
738 ("currency", "usd".to_owned()),
739 ("description", format!("{}: g1t usage", line.workspace)),
740 ("metadata[workspace]", line.workspace.clone()),
741 ];
742 let _: Value = stripe.post_idempotent("/invoiceitems", &fields, &format!("{key}/item/{}", line.workspace)).await?;
743 }
744 // A retry finds it finalized already; that is fine.
745 let _ = stripe.post::<Value>(&format!("/invoices/{}/finalize", draft.id), &[]).await;
746 let sent: Invoice = stripe.post(&format!("/invoices/{}/send", draft.id), &[]).await?;
747 let now = rfc3339(now_ms());
748 let total = lines.iter().map(|l| l.amount_micros).sum::<i64>().max(sent.amount_due * 10_000);
749 let mut writes = vec![self
750 .db
751 .prepare(
752 "INSERT INTO enterprise_invoices (invoice_id, account_id, period, amount_micros, status, hosted_url, created_by, created_at)
753 VALUES (?, ?, ?, ?, 'open', ?, ?, ?)",
754 )
755 .bind(&[
756 sent.id.as_str().into(),
757 id.into(),
758 period.into(),
759 (total as f64).into(),
760 crate::optional(sent.hosted_invoice_url.as_deref()),
761 by.into(),
762 now.as_str().into(),
763 ])?];
764 for line in &lines {
765 writes.push(
766 self.db
767 .prepare("INSERT INTO enterprise_invoice_lines (invoice_id, workspace, amount_micros) VALUES (?, ?, ?)")
768 .bind(&[sent.id.as_str().into(), line.workspace.as_str().into(), (line.amount_micros as f64).into()])?,
769 );
770 }
771 self.db.batch(writes).await?;
772 self.audit(id, "invoice", &format!("Invoice {} for {} sent ({period})", sent.id, crate::features::dollars(total)), by)
773 .await?;
774 Ok(Ok(EnterpriseInvoice {
775 invoice_id: sent.id,
776 hosted_url: sent.hosted_invoice_url,
777 amount_micros: total,
778 status: "open".to_owned(),
779 period: period.to_owned(),
780 lines,
781 created_at: now,
782 }))
783 }
784
785 /// An enterprise invoice paid: each workspace is credited its line, and
786 /// any stop for the invoice is lifted.
787 async fn enterprise_invoice_paid(&self, invoice_id: &str) -> Result<String> {
788 let claimed = self
789 .db
790 .prepare(
791 "UPDATE enterprise_invoices SET status = 'paid', paid_at = ? WHERE invoice_id = ? AND status <> 'paid'
792 RETURNING invoice_id",
793 )
794 .bind(&[rfc3339(now_ms()).into(), invoice_id.into()])?
795 .first::<Value>(None)
796 .await?;
797 if claimed.is_none() {
798 return Ok("ignored: not an open enterprise invoice".to_owned());
799 }
800 let lines = self.invoice_lines(invoice_id).await?;
801 for line in &lines {
802 self.enter(
803 &line.workspace,
804 EntryKind::TopUp,
805 line.amount_micros,
806 &format!("Paid on the enterprise's invoice {invoice_id}"),
807 &format!("inv/{invoice_id}/{}", line.workspace),
808 None,
809 None,
810 None,
811 None,
812 )
813 .await?;
814 }
815 Ok(format!("invoice {invoice_id} paid; {} workspaces credited", lines.len()))
816 }
817
818 /// An enterprise invoice that went overdue stops its workspaces' work;
819 /// one voided is simply closed.
820 async fn enterprise_invoice_status(&self, invoice_id: &str, status: &str) -> Result<String> {
821 let updated = self
822 .db
823 .prepare("UPDATE enterprise_invoices SET status = ? WHERE invoice_id = ? AND status <> 'paid' RETURNING invoice_id")
824 .bind(&[status.into(), invoice_id.into()])?
825 .first::<Value>(None)
826 .await?;
827 if updated.is_none() {
828 return Ok("ignored: not an open enterprise invoice".to_owned());
829 }
830 if status == "overdue" {
831 let now = rfc3339(now_ms());
832 for line in self.invoice_lines(invoice_id).await? {
833 self.db
834 .prepare(
835 "INSERT INTO limits (workspace, autopay_failed_at, autopay_error, updated_at) VALUES (?1, ?2, ?3, ?2)
836 ON CONFLICT (workspace) DO UPDATE SET autopay_failed_at = ?2, autopay_error = ?3, updated_at = ?2",
837 )
838 .bind(&[line.workspace.as_str().into(), now.as_str().into(), format!("the enterprise's invoice {invoice_id} is overdue").into()])?
839 .run()
840 .await?;
841 }
842 }
843 Ok(format!("invoice {invoice_id} is {status}"))
844 }
845
846 async fn invoice_lines(&self, invoice_id: &str) -> Result<Vec<InvoiceLine>> {
847 Ok(self
848 .db
849 .prepare("SELECT workspace, amount_micros FROM enterprise_invoice_lines WHERE invoice_id = ?")
850 .bind(&[invoice_id.into()])?
851 .all()
852 .await?
853 .results::<LineRow>()?
854 .into_iter()
855 .map(|row| InvoiceLine { workspace: row.workspace, amount_micros: row.amount_micros })
856 .collect())
857 }
858
859 /// An enterprise's invoices, newest first.
860 pub(crate) async fn enterprise_invoices(&self, id: &str) -> Result<Vec<EnterpriseInvoice>> {
861 let rows = self
862 .db
863 .prepare("SELECT * FROM enterprise_invoices WHERE account_id = ? ORDER BY created_at DESC LIMIT 24")
864 .bind(&[JsValue::from(id)])?
865 .all()
866 .await?
867 .results::<InvoiceRow>()?;
868 let mut invoices = vec![];
869 for row in rows {
870 invoices.push(EnterpriseInvoice {
871 lines: self.invoice_lines(&row.invoice_id).await?,
872 invoice_id: row.invoice_id,
873 hosted_url: row.hosted_url,
874 amount_micros: row.amount_micros,
875 status: row.status,
876 period: row.period,
877 created_at: row.created_at,
878 });
879 }
880 Ok(invoices)
881 }
882}
883
884#[cfg(test)]
885mod tests {
886 use super::*;
887
888 #[test]
889 fn an_enterprise_invoice_attempted_again_is_the_same_invoice() {
890 let lines = [("acme", 1_250), ("acme-labs", 99)];
891 let key = enterprise_invoice_key("ent_1", "2026-09", &lines);
892 assert_eq!(key, enterprise_invoice_key("ent_1", "2026-09", &lines));
893 assert!(key.len() < 255 && key.starts_with("ent-invoice/ent_1/2026-09/"));
894 // Different amounts, even with the same total, are a different invoice.
895 assert_ne!(key, enterprise_invoice_key("ent_1", "2026-09", &[("acme", 1_300), ("acme-labs", 49)]));
896 assert_ne!(key, enterprise_invoice_key("ent_1", "2026-10", &lines));
897 }
898
899 fn sign(payload: &str, secret: &str, t: i64) -> String {
900 let mut mac = Hmac::<Sha256>::new_from_slice(secret.as_bytes()).unwrap();
901 mac.update(format!("{t}.{payload}").as_bytes());
902 format!("t={t},v1={}", hex::encode(mac.finalize().into_bytes()))
903 }
904
905 #[test]
906 fn a_destination_misses_the_events_billing_handles_that_it_does_not_send() {
907 let has: Vec<String> = EVENTS.iter().take(11).map(|e| (*e).to_owned()).collect();
908 assert_eq!(missing_events(&has), EVENTS[11..].iter().map(|e| (*e).to_owned()).collect::<Vec<_>>());
909 let all: Vec<String> = EVENTS.iter().map(|e| (*e).to_owned()).collect();
910 assert!(missing_events(&all).is_empty());
911 assert!(missing_events(&["*".to_owned()]).is_empty());
912 }
913
914 #[test]
915 fn a_signed_event_is_believed_only_as_signed_and_only_fresh() {
916 let payload = r#"{"id":"evt_1","type":"invoice.paid"}"#;
917 let header = sign(payload, "whsec_test", 1_000_000);
918 assert!(verify(payload, &header, "whsec_test", 1_000_010));
919 assert!(!verify(payload, &header, "whsec_other", 1_000_010));
920 assert!(!verify(&payload.replace("paid", "voided"), &header, "whsec_test", 1_000_010));
921 assert!(!verify(payload, &header, "whsec_test", 1_000_000 + 301));
922 assert!(!verify(payload, "v1=abc", "whsec_test", 1_000_000));
923 // Stripe may sign with more than one secret while one is rolled.
924 let both = format!("{},v1=00ff", sign(payload, "whsec_test", 1_000_000));
925 assert!(verify(payload, &both, "whsec_test", 1_000_000));
926 }
927}