Sign in

pulls/pr_01m47d24b0e6n91zwymwxg0vpx

public

pr_01m47d24b0e6n91zwymwxg0vpx/services/billing/src/webhooks.rs

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