Sign in

pulls/pr_01m47d15m3e54sn21z27rpy5n9

public

pr_01m47d15m3e54sn21z27rpy5n9/services/billing/src/webhooks.rs

809 lines34,139 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 {
277 self.enterprise_invoice_paid(&text("id")).await?
278 }
279 }
280 "invoice.payment_failed" => match object["subscription"].as_str() {
281 Some(subscription) => self.settle_subscription(subscription).await?,
282 None => "ignored: not a plan".to_owned(),
283 },
284 "invoice.overdue" => self.enterprise_invoice_status(&text("id"), "overdue").await?,
285 "invoice.voided" => self.enterprise_invoice_status(&text("id"), "void").await?,
286 "charge.refunded" => self.refunded(object).await?,
287 "charge.dispute.created" => self.disputed(object, true).await?,
288 "charge.dispute.closed" => self.disputed(object, object["status"].as_str() == Some("lost")).await?,
289 _ => "ignored".to_owned(),
290 })
291 }
292
293 /// A payment page done, whether or not the person came back to g1t.
294 async fn settle_checkout(&self, session_id: &str) -> Result<String> {
295 #[derive(Deserialize)]
296 struct Open {
297 workspace: String,
298 created_by: String,
299 feature: Option<String>,
300 }
301 let Some(open) = self
302 .db
303 .prepare("SELECT workspace, created_by, feature FROM checkouts WHERE id = ? AND status = 'open'")
304 .bind(&[session_id.into()])?
305 .first::<Open>(None)
306 .await?
307 else {
308 return Ok("ignored: already settled or not g1t's".to_owned());
309 };
310 let Some(stripe) = &self.stripe else { return Ok("ignored: payments off".to_owned()) };
311 let session = stripe.session(session_id).await?;
312 if session.payment_status != "paid" && open.feature.is_none() {
313 return Ok("ignored: not paid".to_owned());
314 }
315 let claimed = self
316 .db
317 .prepare("UPDATE checkouts SET status = 'paid' WHERE id = ? AND status = 'open' RETURNING id")
318 .bind(&[session_id.into()])?
319 .first::<Value>(None)
320 .await?;
321 if claimed.is_none() {
322 return Ok("ignored: settled meanwhile".to_owned());
323 }
324 match open.feature.as_deref() {
325 None => {
326 let cents = i64::from(session.amount_total.unwrap_or(0));
327 self.enter(
328 &open.workspace,
329 EntryKind::TopUp,
330 cents * 10_000,
331 "Credit added by card",
332 &session.id,
333 None,
334 None,
335 Some(&open.created_by),
336 session.customer.as_deref(),
337 )
338 .await?;
339 Ok(format!("credited {} to {}", crate::features::dollars(cents * 10_000), open.workspace))
340 }
341 Some(feature) => {
342 let Some(feature) = g1t_contracts::billing::Feature::parse(feature) else {
343 return Ok("ignored: unknown feature".to_owned());
344 };
345 if let Some(subscription_id) = &session.subscription {
346 let subscription = stripe.subscription(subscription_id).await?;
347 self.record(&open.workspace, feature, &subscription, &open.created_by).await?;
348 }
349 self.db
350 .prepare(
351 "INSERT INTO accounts (workspace, balance_micros, customer_id, created_at) VALUES (?1, 0, ?2, ?3)
352 ON CONFLICT (workspace) DO UPDATE SET customer_id = COALESCE(customer_id, ?2)",
353 )
354 .bind(&[open.workspace.as_str().into(), crate::optional(session.customer.as_deref()), rfc3339(now_ms()).into()])?
355 .run()
356 .await?;
357 Ok(format!("{} plan started for {}", feature.title(), open.workspace))
358 }
359 }
360 }
361
362 /// A plan that changed at Stripe: renewed, failed, canceled.
363 async fn settle_subscription(&self, subscription_id: &str) -> Result<String> {
364 #[derive(Deserialize)]
365 struct Plan {
366 workspace: String,
367 feature: String,
368 started_by: String,
369 }
370 let Some(plan) = self
371 .db
372 .prepare("SELECT workspace, feature, started_by FROM subscriptions WHERE subscription_id = ?")
373 .bind(&[subscription_id.into()])?
374 .first::<Plan>(None)
375 .await?
376 else {
377 return Ok("ignored: not a g1t plan".to_owned());
378 };
379 let (Some(stripe), Some(feature)) = (&self.stripe, g1t_contracts::billing::Feature::parse(&plan.feature)) else {
380 return Ok("ignored".to_owned());
381 };
382 let subscription = stripe.subscription(subscription_id).await?;
383 self.record(&plan.workspace, feature, &subscription, &plan.started_by).await?;
384 Ok(format!("{} plan for {} is {}", feature.title(), plan.workspace, subscription.status))
385 }
386
387 /// The workspace a Stripe customer belongs to.
388 async fn workspace_of_customer(&self, customer: &str) -> Result<Option<String>> {
389 #[derive(Deserialize)]
390 struct Row {
391 workspace: String,
392 }
393 Ok(self
394 .db
395 .prepare("SELECT workspace FROM accounts WHERE customer_id = ?")
396 .bind(&[customer.into()])?
397 .first::<Row>(None)
398 .await?
399 .map(|row| row.workspace))
400 }
401
402 /// Money given back: what was paid is less, by the refund.
403 async fn refunded(&self, charge: &Value) -> Result<String> {
404 let Some(customer) = charge["customer"].as_str() else { return Ok("ignored: no customer".to_owned()) };
405 let Some(workspace) = self.workspace_of_customer(customer).await? else {
406 return Ok("ignored: not a workspace's customer".to_owned());
407 };
408 let refunded = charge["amount_refunded"].as_i64().unwrap_or(0);
409 if refunded <= 0 {
410 return Ok("ignored: nothing refunded".to_owned());
411 }
412 // Each refund total once, so partial refunds add up correctly.
413 let charge_id = charge["id"].as_str().unwrap_or_default();
414 #[derive(Deserialize)]
415 struct Sum {
416 micros: Option<i64>,
417 }
418 let already = self
419 .db
420 .prepare("SELECT -SUM(amount_micros) AS micros FROM ledger WHERE reference LIKE ?")
421 .bind(&[format!("refund/{charge_id}/%").into()])?
422 .first::<Sum>(None)
423 .await?
424 .and_then(|s| s.micros)
425 .unwrap_or(0);
426 let new = refunded * 10_000 - already;
427 if new <= 0 {
428 return Ok("ignored: refund already recorded".to_owned());
429 }
430 self.enter(
431 &workspace,
432 EntryKind::TopUp,
433 -new,
434 "Refunded to the card",
435 &format!("refund/{charge_id}/{refunded}"),
436 None,
437 None,
438 None,
439 None,
440 )
441 .await?;
442 Ok(format!("refund of {} recorded for {workspace}", crate::features::dollars(new)))
443 }
444
445 /// A disputed payment stops the workspace's work until it is resolved;
446 /// one closed in the workspace's favour lets it go on.
447 async fn disputed(&self, dispute: &Value, stop: bool) -> Result<String> {
448 let Some(stripe) = &self.stripe else { return Ok("ignored".to_owned()) };
449 let Some(charge_id) = dispute["charge"].as_str() else { return Ok("ignored: no charge".to_owned()) };
450 let charge: Value = stripe.get(&format!("/charges/{charge_id}")).await?;
451 let Some(workspace) = (match charge["customer"].as_str() {
452 Some(customer) => self.workspace_of_customer(customer).await?,
453 None => None,
454 }) else {
455 return Ok("ignored: not a workspace's customer".to_owned());
456 };
457 let now = rfc3339(now_ms());
458 if stop {
459 self.db
460 .prepare(
461 "INSERT INTO limits (workspace, autopay_failed_at, autopay_error, updated_at) VALUES (?1, ?2, ?3, ?2)
462 ON CONFLICT (workspace) DO UPDATE SET autopay_failed_at = ?2, autopay_error = ?3, updated_at = ?2",
463 )
464 .bind(&[workspace.as_str().into(), now.as_str().into(), "a payment was disputed with the card's bank".into()])?
465 .run()
466 .await?;
467 } else {
468 self.db
469 .prepare("UPDATE limits SET autopay_failed_at = NULL, autopay_error = NULL WHERE workspace = ?")
470 .bind(&[workspace.as_str().into()])?
471 .run()
472 .await?;
473 }
474 let account = self.account_of(&workspace).await?;
475 let what = if stop { "dispute: work stopped" } else { "dispute closed in the workspace's favour" };
476 self.audit(&account.id, "dispute", &format!("{workspace}: {what}"), "stripe").await?;
477 Ok(format!("{workspace}: {what}"))
478 }
479
480 // --- Enterprise invoices ------------------------------------------------
481
482 pub(crate) async fn admin_enterprise_billing(&self, a: AdminEnterpriseBillingArgs) -> Result<Outcome<BillingAccount>> {
483 let email = a.email.trim().to_lowercase();
484 if !email.contains('@') || email.contains(char::is_whitespace) || a.by.trim().is_empty() {
485 return Ok(Outcome::fail(FailureCode::Invalid, "Give the email the enterprise's invoices go to."));
486 }
487 let Some(stripe) = &self.stripe else {
488 return Ok(Outcome::fail(FailureCode::Conflict, "Payments are not set up on this g1t."));
489 };
490 #[derive(Deserialize)]
491 struct Row {
492 name: String,
493 kind: String,
494 customer_id: Option<String>,
495 }
496 let Some(row) = self
497 .db
498 .prepare("SELECT name, kind, customer_id FROM billing_accounts WHERE id = ?")
499 .bind(&[a.id.as_str().into()])?
500 .first::<Row>(None)
501 .await?
502 .filter(|row| row.kind == "enterprise")
503 else {
504 return Ok(Outcome::fail(FailureCode::NotFound, "No such enterprise."));
505 };
506 #[derive(Deserialize)]
507 struct Customer {
508 id: String,
509 }
510 let fields = [
511 ("name", row.name.clone()),
512 ("email", email.clone()),
513 ("metadata[g1t_enterprise]", a.id.clone()),
514 ];
515 let customer: Customer = match &row.customer_id {
516 Some(id) => stripe.post(&format!("/customers/{id}"), &fields).await?,
517 None => stripe.post("/customers", &fields).await?,
518 };
519 self.db
520 .prepare("UPDATE billing_accounts SET billing_email = ?, customer_id = ? WHERE id = ?")
521 .bind(&[email.as_str().into(), customer.id.as_str().into(), a.id.as_str().into()])?
522 .run()
523 .await?;
524 self.audit(&a.id, "billing_email", &format!("Invoices go to {email}"), &a.by).await?;
525 Ok(match self.enterprise(&a.id).await? {
526 Some(account) => Outcome::Ok(account),
527 None => Outcome::fail(FailureCode::NotFound, "No such enterprise."),
528 })
529 }
530
531 pub(crate) async fn admin_invoice_enterprise(&self, a: AdminInvoiceEnterpriseArgs) -> Result<Outcome<EnterpriseInvoice>> {
532 if a.by.trim().is_empty() {
533 return Ok(Outcome::fail(FailureCode::Invalid, "Say who is sending it."));
534 }
535 match self.invoice_enterprise(&a.id, "now", &a.by).await? {
536 Ok(invoice) => Ok(Outcome::Ok(invoice)),
537 Err(why) => Ok(Outcome::fail(FailureCode::Conflict, why)),
538 }
539 }
540
541 /// Invoices each enterprise for the month that closed. Live payments
542 /// only; sudo can send one sooner in test mode.
543 pub(crate) async fn invoice_enterprises(&self) -> Result<()> {
544 if !self.stripe.as_ref().is_some_and(crate::stripe::Stripe::live) {
545 return Ok(());
546 }
547 let closing = crate::limits::previous_month(&rfc3339(now_ms())[..7]);
548 #[derive(Deserialize)]
549 struct Id {
550 id: String,
551 }
552 let due = self
553 .db
554 .prepare(
555 "SELECT id FROM billing_accounts WHERE kind = 'enterprise' AND customer_id IS NOT NULL
556 AND terms_kind <> 'comped'
557 AND NOT EXISTS (SELECT 1 FROM enterprise_invoices i WHERE i.account_id = billing_accounts.id AND i.period = ?)",
558 )
559 .bind(&[closing.as_str().into()])?
560 .all()
561 .await?
562 .results::<Id>()?;
563 for Id { id } in due {
564 if let Err(why) = self.invoice_enterprise(&id, &closing, "month close").await? {
565 worker::console_log!("enterprise {id} not invoiced for {closing}: {why}");
566 }
567 }
568 Ok(())
569 }
570
571 /// One invoice for what each of the enterprise's workspaces owes now.
572 async fn invoice_enterprise(&self, id: &str, period: &str, by: &str) -> Result<std::result::Result<EnterpriseInvoice, String>> {
573 let Some(stripe) = &self.stripe else { return Ok(Err("Payments are not set up.".into())) };
574 let Some(account) = self.enterprise(id).await? else { return Ok(Err("No such enterprise.".into())) };
575 #[derive(Deserialize)]
576 struct Customer {
577 customer_id: Option<String>,
578 }
579 let Some(customer) = self
580 .db
581 .prepare("SELECT customer_id FROM billing_accounts WHERE id = ?")
582 .bind(&[id.into()])?
583 .first::<Customer>(None)
584 .await?
585 .and_then(|row| row.customer_id)
586 else {
587 return Ok(Err("Set where the enterprise's invoices go first.".into()));
588 };
589 // What each workspace owes: its charges less what it has paid, and
590 // less what is on invoices still open.
591 let mut lines = vec![];
592 for workspace in &account.workspaces {
593 let balance = self.row(workspace).await?.map_or(0, |row| row.balance_micros);
594 #[derive(Deserialize)]
595 struct Sum {
596 micros: Option<i64>,
597 }
598 let invoiced = self
599 .db
600 .prepare(
601 "SELECT SUM(l.amount_micros) AS micros FROM enterprise_invoice_lines l
602 JOIN enterprise_invoices i ON i.invoice_id = l.invoice_id
603 WHERE l.workspace = ? AND i.status IN ('open', 'overdue')",
604 )
605 .bind(&[workspace.as_str().into()])?
606 .first::<Sum>(None)
607 .await?
608 .and_then(|s| s.micros)
609 .unwrap_or(0);
610 let owed = (-balance).max(0) - invoiced;
611 if owed >= 10_000 {
612 lines.push(InvoiceLine { workspace: workspace.clone(), amount_micros: owed });
613 }
614 }
615 if lines.is_empty() {
616 return Ok(Err("Its workspaces owe nothing to invoice.".into()));
617 }
618 for line in &lines {
619 let cents = (line.amount_micros + 9_999) / 10_000;
620 let fields = [
621 ("customer", customer.clone()),
622 ("amount", cents.to_string()),
623 ("currency", "usd".to_owned()),
624 ("description", format!("{}: g1t usage", line.workspace)),
625 ("metadata[workspace]", line.workspace.clone()),
626 ];
627 let _: Value = stripe.post("/invoiceitems", &fields).await?;
628 }
629 let fields = [
630 ("customer", customer.clone()),
631 ("collection_method", "send_invoice".to_owned()),
632 ("days_until_due", "30".to_owned()),
633 ("pending_invoice_items_behavior", "include".to_owned()),
634 ("description", format!("g1t usage for the {} enterprise", account.name)),
635 ("metadata[g1t_enterprise]", id.to_owned()),
636 ("metadata[period]", period.to_owned()),
637 ];
638 #[derive(Deserialize)]
639 struct Invoice {
640 id: String,
641 #[serde(default)]
642 hosted_invoice_url: Option<String>,
643 #[serde(default)]
644 amount_due: i64,
645 }
646 let draft: Invoice = stripe.post("/invoices", &fields).await?;
647 let _: Value = stripe.post(&format!("/invoices/{}/finalize", draft.id), &[]).await?;
648 let sent: Invoice = stripe.post(&format!("/invoices/{}/send", draft.id), &[]).await?;
649 let now = rfc3339(now_ms());
650 let total = lines.iter().map(|l| l.amount_micros).sum::<i64>().max(sent.amount_due * 10_000);
651 let mut writes = vec![self
652 .db
653 .prepare(
654 "INSERT INTO enterprise_invoices (invoice_id, account_id, period, amount_micros, status, hosted_url, created_by, created_at)
655 VALUES (?, ?, ?, ?, 'open', ?, ?, ?)",
656 )
657 .bind(&[
658 sent.id.as_str().into(),
659 id.into(),
660 period.into(),
661 (total as f64).into(),
662 crate::optional(sent.hosted_invoice_url.as_deref()),
663 by.into(),
664 now.as_str().into(),
665 ])?];
666 for line in &lines {
667 writes.push(
668 self.db
669 .prepare("INSERT INTO enterprise_invoice_lines (invoice_id, workspace, amount_micros) VALUES (?, ?, ?)")
670 .bind(&[sent.id.as_str().into(), line.workspace.as_str().into(), (line.amount_micros as f64).into()])?,
671 );
672 }
673 self.db.batch(writes).await?;
674 self.audit(id, "invoice", &format!("Invoice {} for {} sent ({period})", sent.id, crate::features::dollars(total)), by)
675 .await?;
676 Ok(Ok(EnterpriseInvoice {
677 invoice_id: sent.id,
678 hosted_url: sent.hosted_invoice_url,
679 amount_micros: total,
680 status: "open".to_owned(),
681 period: period.to_owned(),
682 lines,
683 created_at: now,
684 }))
685 }
686
687 /// An enterprise invoice paid: each workspace is credited its line, and
688 /// any stop for the invoice is lifted.
689 async fn enterprise_invoice_paid(&self, invoice_id: &str) -> Result<String> {
690 let claimed = self
691 .db
692 .prepare(
693 "UPDATE enterprise_invoices SET status = 'paid', paid_at = ? WHERE invoice_id = ? AND status <> 'paid'
694 RETURNING invoice_id",
695 )
696 .bind(&[rfc3339(now_ms()).into(), invoice_id.into()])?
697 .first::<Value>(None)
698 .await?;
699 if claimed.is_none() {
700 return Ok("ignored: not an open enterprise invoice".to_owned());
701 }
702 let lines = self.invoice_lines(invoice_id).await?;
703 for line in &lines {
704 self.enter(
705 &line.workspace,
706 EntryKind::TopUp,
707 line.amount_micros,
708 &format!("Paid on the enterprise's invoice {invoice_id}"),
709 &format!("inv/{invoice_id}/{}", line.workspace),
710 None,
711 None,
712 None,
713 None,
714 )
715 .await?;
716 }
717 Ok(format!("invoice {invoice_id} paid; {} workspaces credited", lines.len()))
718 }
719
720 /// An enterprise invoice that went overdue stops its workspaces' work;
721 /// one voided is simply closed.
722 async fn enterprise_invoice_status(&self, invoice_id: &str, status: &str) -> Result<String> {
723 let updated = self
724 .db
725 .prepare("UPDATE enterprise_invoices SET status = ? WHERE invoice_id = ? AND status <> 'paid' RETURNING invoice_id")
726 .bind(&[status.into(), invoice_id.into()])?
727 .first::<Value>(None)
728 .await?;
729 if updated.is_none() {
730 return Ok("ignored: not an open enterprise invoice".to_owned());
731 }
732 if status == "overdue" {
733 let now = rfc3339(now_ms());
734 for line in self.invoice_lines(invoice_id).await? {
735 self.db
736 .prepare(
737 "INSERT INTO limits (workspace, autopay_failed_at, autopay_error, updated_at) VALUES (?1, ?2, ?3, ?2)
738 ON CONFLICT (workspace) DO UPDATE SET autopay_failed_at = ?2, autopay_error = ?3, updated_at = ?2",
739 )
740 .bind(&[line.workspace.as_str().into(), now.as_str().into(), format!("the enterprise's invoice {invoice_id} is overdue").into()])?
741 .run()
742 .await?;
743 }
744 }
745 Ok(format!("invoice {invoice_id} is {status}"))
746 }
747
748 async fn invoice_lines(&self, invoice_id: &str) -> Result<Vec<InvoiceLine>> {
749 Ok(self
750 .db
751 .prepare("SELECT workspace, amount_micros FROM enterprise_invoice_lines WHERE invoice_id = ?")
752 .bind(&[invoice_id.into()])?
753 .all()
754 .await?
755 .results::<LineRow>()?
756 .into_iter()
757 .map(|row| InvoiceLine { workspace: row.workspace, amount_micros: row.amount_micros })
758 .collect())
759 }
760
761 /// An enterprise's invoices, newest first.
762 pub(crate) async fn enterprise_invoices(&self, id: &str) -> Result<Vec<EnterpriseInvoice>> {
763 let rows = self
764 .db
765 .prepare("SELECT * FROM enterprise_invoices WHERE account_id = ? ORDER BY created_at DESC LIMIT 24")
766 .bind(&[JsValue::from(id)])?
767 .all()
768 .await?
769 .results::<InvoiceRow>()?;
770 let mut invoices = vec![];
771 for row in rows {
772 invoices.push(EnterpriseInvoice {
773 lines: self.invoice_lines(&row.invoice_id).await?,
774 invoice_id: row.invoice_id,
775 hosted_url: row.hosted_url,
776 amount_micros: row.amount_micros,
777 status: row.status,
778 period: row.period,
779 created_at: row.created_at,
780 });
781 }
782 Ok(invoices)
783 }
784}
785
786#[cfg(test)]
787mod tests {
788 use super::*;
789
790 fn sign(payload: &str, secret: &str, t: i64) -> String {
791 let mut mac = Hmac::<Sha256>::new_from_slice(secret.as_bytes()).unwrap();
792 mac.update(format!("{t}.{payload}").as_bytes());
793 format!("t={t},v1={}", hex::encode(mac.finalize().into_bytes()))
794 }
795
796 #[test]
797 fn a_signed_event_is_believed_only_as_signed_and_only_fresh() {
798 let payload = r#"{"id":"evt_1","type":"invoice.paid"}"#;
799 let header = sign(payload, "whsec_test", 1_000_000);
800 assert!(verify(payload, &header, "whsec_test", 1_000_010));
801 assert!(!verify(payload, &header, "whsec_other", 1_000_010));
802 assert!(!verify(&payload.replace("paid", "voided"), &header, "whsec_test", 1_000_010));
803 assert!(!verify(payload, &header, "whsec_test", 1_000_000 + 301));
804 assert!(!verify(payload, "v1=abc", "whsec_test", 1_000_000));
805 // Stripe may sign with more than one secret while one is rolled.
806 let both = format!("{},v1=00ff", sign(payload, "whsec_test", 1_000_000));
807 assert!(verify(payload, &both, "whsec_test", 1_000_000));
808 }
809}