Sign in

flagon-io/g1t

public

g1t/services/billing/src/webhooks.rs

895 lines38,316 bytesCodeBlame

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

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