Skip to content
Sign in

flagon-io/g1t

public
0

g1t/services/billing/src/webhooks.rs

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