Sign in

pulls/pr_01m47d24b0e6n91zwymwxg0vpx

public

pr_01m47d24b0e6n91zwymwxg0vpx/services/billing/src/webhooks.rs

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