Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh
gitagentscloudflarerustforgemcp
406 linesCodeBlameRaw
1//! What billing keeps of Stripe, so that reads never wait on it and nothing
2//! Stripe says is lost:
3//!
4//! - The saved card, on the account row: read from there, asked of Stripe
5//! only the first time, and refreshed by card and customer events
6//! (webhooks.rs) and a weekly pass.
7//! - Missed events, replayed: every run of the cron reads Stripe's event
8//! list from where the last run got to and handles any event not seen,
9//! through the same once-only claim the webhook uses. A delivery that
10//! failed for good, or an endpoint not registered yet, costs at most one
11//! cron interval.
12//! - The destination, kept: once a day it is enabled again if Stripe turned
13//! it off after failures, and given any event billing handles that it
14//! does not send. Its signing secret, `STRIPE_WEBHOOK_SECRET`, is not
15//! touched.
16
17use crate::Billing;
18use crate::stripe::form;
19use crate::webhooks::{EVENTS, WEBHOOK_URL, missing_events};
20use g1t_contracts::billing::Card;
21use g1t_contracts::time::rfc3339;
22use g1t_kit::now_ms;
23use serde::Deserialize;
24use serde_json::Value;
25use worker::Result;
26
27/// How far back each replay starts before where the last one got to, for
28/// events Stripe lists a little late.
29const OVERLAP_SECONDS: i64 = 60 * 60;
30/// How recent an event may be and still be listed later: the replay's
31/// mark stays this far behind the clock.
32const SETTLE_SECONDS: i64 = 5 * 60;
33/// Where the first replay starts.
34const FIRST_LOOK_SECONDS: i64 = 3 * 24 * 60 * 60;
35/// Pages of 100 events one replay reads at most.
36const MAX_PAGES: usize = 10;
37/// A claim this old that never finished: its handler died, so the event
38/// may be handled again.
39const STALE_CLAIM_MS: u64 = 10 * 60 * 1000;
40/// How long a saved card is believed before the weekly pass asks again.
41const CARD_FRESH_DAYS: u64 = 7;
42/// Accounts or plans the daily pass refreshes at most.
43const REFRESH_BATCH: u32 = 25;
44
45#[derive(Deserialize)]
46struct CardRow {
47 customer_id: Option<String>,
48 card_brand: Option<String>,
49 card_last4: Option<String>,
50 card_exp_month: Option<u32>,
51 card_exp_year: Option<u32>,
52 card_synced_at: Option<String>,
53}
54
55/// Where the replay's mark goes after a run: the first event that failed
56/// (so it is listed again), the old mark when the list was not read to its
57/// end, else just behind the clock.
58pub(crate) fn next_through(previous: i64, now: i64, first_failed: Option<i64>, complete: bool) -> i64 {
59 match first_failed {
60 Some(created) if !complete => previous.min(created),
61 Some(created) => created.min(previous.max(created)),
62 None if !complete => previous,
63 None => previous.max(now - SETTLE_SECONDS),
64 }
65}
66
67/// A value to bind, or SQL NULL. `Option::into` gives JavaScript
68/// `undefined` for None, which D1 refuses ("Type 'undefined' not
69/// supported"): a customer without a card, or a replay that finished.
70fn or_null<T: Into<worker::wasm_bindgen::JsValue>>(value: Option<T>) -> worker::wasm_bindgen::JsValue {
71 value.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into)
72}
73
74impl Billing {
75 /// The workspace's saved card as last synced. Stripe is asked only when
76 /// it never was, or g1t changed the card since; a Stripe that does not
77 /// answer then shows no card, as before.
78 pub(crate) async fn saved_card(&self, workspace: &str) -> Result<Option<Card>> {
79 let Some(row) = self
80 .db
81 .prepare(
82 "SELECT customer_id, card_brand, card_last4, card_exp_month, card_exp_year, card_synced_at
83 FROM accounts WHERE workspace = ?",
84 )
85 .bind(&[workspace.into()])?
86 .first::<CardRow>(None)
87 .await?
88 else {
89 return Ok(None);
90 };
91 if row.card_synced_at.is_some() {
92 return Ok(card_of(&row));
93 }
94 let Some(customer) = row.customer_id.as_deref() else { return Ok(None) };
95 if self.stripe.is_none() {
96 return Ok(None);
97 }
98 Ok(match self.fetch_card(customer).await {
99 Ok(card) => card,
100 Err(error) => {
101 worker::console_error!("reading {workspace}'s card from Stripe failed: {error}");
102 None
103 }
104 })
105 }
106
107 /// Asks Stripe for the customer's card and keeps it on every account
108 /// with that customer.
109 async fn fetch_card(&self, customer: &str) -> Result<Option<Card>> {
110 let Some(stripe) = &self.stripe else { return Ok(None) };
111 // The card invoices are charged to (the default), not merely the
112 // newest: what Stripe's billing page made the default is the one.
113 let method = match stripe.default_payment_method(customer).await {
114 Ok(method) => method,
115 // Deleted at Stripe, or from another account: every account
116 // with it forgets it, so the next card saved makes a new one.
117 Err(error) if crate::stripe::is_missing(&error) => {
118 self.db
119 .prepare(
120 "UPDATE accounts SET customer_id = NULL, card_brand = NULL, card_last4 = NULL, card_exp_month = NULL,
121 card_exp_year = NULL, card_synced_at = NULL WHERE customer_id = ?",
122 )
123 .bind(&[customer.into()])?
124 .run()
125 .await?;
126 return Ok(None);
127 }
128 Err(error) => return Err(error),
129 };
130 let card = method.filter(|m| m.kind == "card").and_then(|m| {
131 Some(Card { brand: m.brand?, last4: m.last4?, exp_month: m.exp_month?, exp_year: m.exp_year? })
132 });
133 self.db
134 .prepare(
135 "UPDATE accounts SET card_brand = ?2, card_last4 = ?3, card_exp_month = ?4, card_exp_year = ?5,
136 card_synced_at = ?6 WHERE customer_id = ?1",
137 )
138 .bind(&[
139 customer.into(),
140 or_null(card.as_ref().map(|c| c.brand.clone())),
141 or_null(card.as_ref().map(|c| c.last4.clone())),
142 or_null(card.as_ref().map(|c| c.exp_month as f64)),
143 or_null(card.as_ref().map(|c| c.exp_year as f64)),
144 rfc3339(now_ms()).into(),
145 ])?
146 .run()
147 .await?;
148 Ok(card)
149 }
150
151 /// A card or customer event: the saved card read again. The event's
152 /// outcome, for `stripe_events`.
153 pub(crate) async fn sync_card_of(&self, customer: &str) -> Result<String> {
154 if self.workspace_of_customer(customer).await?.is_none() {
155 return Ok("ignored: not a workspace's customer".to_owned());
156 }
157 Ok(match self.fetch_card(customer).await? {
158 Some(card) => format!("card saved: {} ending {}", card.brand, card.last4),
159 None => "card saved: none".to_owned(),
160 })
161 }
162
163 /// Forgets the saved card after g1t itself changed it, so the next
164 /// read asks Stripe instead of waiting for the event.
165 pub(crate) async fn forget_card(&self, workspace: &str) -> Result<()> {
166 self.db
167 .prepare("UPDATE accounts SET card_synced_at = NULL WHERE workspace = ?")
168 .bind(&[workspace.into()])?
169 .run()
170 .await?;
171 Ok(())
172 }
173
174 /// Handles events Stripe sent that billing never saw, from its event
175 /// list. What happened, for the log.
176 pub(crate) async fn replay_events(&self) -> Result<String> {
177 let Some(stripe) = &self.stripe else { return Ok("payments are not set up".to_owned()) };
178 #[derive(Deserialize)]
179 struct Mark {
180 through: i64,
181 replay_after: Option<String>,
182 replay_since: Option<i64>,
183 replay_until: Option<i64>,
184 replay_failed: Option<i64>,
185 }
186 #[derive(Deserialize)]
187 struct Page {
188 data: Vec<Value>,
189 has_more: bool,
190 }
191 let mode = self.mode();
192 let now = (now_ms() / 1000) as i64;
193 // A claim whose handler died never finished: let it be handled again.
194 self.db
195 .prepare("DELETE FROM stripe_events WHERE outcome = 'handling' AND received_at < ?")
196 .bind(&[rfc3339(now_ms().saturating_sub(STALE_CLAIM_MS)).into()])?
197 .run()
198 .await?;
199 let mark = self
200 .db
201 .prepare("SELECT through, replay_after, replay_since, replay_until, replay_failed FROM stripe_sync WHERE mode = ?")
202 .bind(&[mode.into()])?
203 .first::<Mark>(None)
204 .await?;
205 let previous = mark.as_ref().map_or(now - FIRST_LOOK_SECONDS, |mark| mark.through);
206 let since = mark.as_ref().and_then(|m| m.replay_since).unwrap_or(previous - OVERLAP_SECONDS);
207 let until = mark.as_ref().and_then(|m| m.replay_until).unwrap_or(now);
208
209 // Stripe lists newest first; read back to `since`.
210 let mut listed: Vec<Value> = Vec::new();
211 let mut complete = false;
212 let mut after = mark.as_ref().and_then(|m| m.replay_after.clone());
213 for _ in 0..MAX_PAGES {
214 let mut query: Vec<(&str, String)> = vec![("limit", "100".to_owned()), ("created[gte]", since.to_string()), ("created[lte]", until.to_string())];
215 query.extend(EVENTS.iter().map(|event| ("types[]", (*event).to_owned())));
216 if let Some(after) = &after {
217 query.push(("starting_after", after.clone()));
218 }
219 let page: Page = stripe.get(&format!("/events?{}", form(&query))).await?;
220 after = page.data.last().and_then(|event| event["id"].as_str()).map(str::to_owned);
221 listed.extend(page.data);
222 if !page.has_more || after.is_none() {
223 complete = true;
224 break;
225 }
226 }
227 if !complete {
228 worker::console_log!("Stripe replay continues from its saved cursor next tick after {} events", MAX_PAGES * 100);
229 }
230
231 // Oldest first, as they happened.
232 listed.sort_by_key(|event| event["created"].as_i64().unwrap_or(0));
233 let mut handled = 0;
234 let mut first_failed = mark.as_ref().and_then(|m| m.replay_failed);
235 for event in &listed {
236 match self.process_event(event).await {
237 Ok(true) => handled += 1,
238 Ok(false) => {}
239 Err(error) => {
240 let id = event["id"].as_str().unwrap_or_default();
241 worker::console_error!("replaying Stripe event {id} failed: {error}");
242 if let Some(created) = event["created"].as_i64() {
243 first_failed = Some(first_failed.map_or(created, |known| known.min(created)));
244 }
245 }
246 }
247 }
248 let through = next_through(previous, until, first_failed, complete);
249 self.db
250 .prepare(
251 "INSERT INTO stripe_sync (mode, through, checked_at, replay_after, replay_since, replay_until, replay_failed)
252 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
253 ON CONFLICT (mode) DO UPDATE SET through = ?2, checked_at = ?3,
254 replay_after = ?4, replay_since = ?5, replay_until = ?6, replay_failed = ?7",
255 )
256 .bind(&[
257 mode.into(), (through as f64).into(), rfc3339(now_ms()).into(),
258 crate::optional(if complete { None } else { after.as_deref() }),
259 or_null((!complete).then_some(since as f64)),
260 or_null((!complete).then_some(until as f64)),
261 or_null(if complete { None } else { first_failed.map(|v| v as f64) }),
262 ])?
263 .run()
264 .await?;
265 Ok(format!("{} listed, {handled} not seen before and handled", listed.len()))
266 }
267
268 /// Keeps the destination at billing's address enabled and sending
269 /// every event billing handles. Its signing secret is not touched.
270 /// Nothing when Stripe has no destination there.
271 pub(crate) async fn keep_endpoint(&self, by: &str) -> Result<String> {
272 let (Some(stripe), Some(destination)) = (&self.stripe, self.destination().await?) else {
273 return Ok(format!("no destination at {WEBHOOK_URL}"));
274 };
275 let path = format!("/webhook_endpoints/{}", destination.id);
276 let mut fields: Vec<(String, String)> = Vec::new();
277 let mut changes = Vec::new();
278 if destination.status != "enabled" {
279 fields.push(("disabled".to_owned(), "false".to_owned()));
280 changes.push("enabled again".to_owned());
281 }
282 let missing = missing_events(&destination.enabled_events);
283 if !missing.is_empty() {
284 // Stripe replaces the list: what it sends now, plus what is missing.
285 let all = destination.enabled_events.iter().cloned().chain(missing.iter().cloned());
286 fields.extend(all.enumerate().map(|(i, event)| (format!("enabled_events[{i}]"), event)));
287 changes.push(format!("added {}", missing.join(", ")));
288 }
289 if changes.is_empty() {
290 return Ok("destination as it should be".to_owned());
291 }
292 let fields: Vec<(&str, String)> = fields.iter().map(|(name, value)| (name.as_str(), value.clone())).collect();
293 let _: Value = stripe.post(&path, &fields).await?;
294 let done = changes.join("; ");
295 self.audit("stripe", "webhook", &format!("Destination {}: {done}", destination.id), by).await?;
296 Ok(done)
297 }
298
299 /// Reads again the saved cards and plans not read in a while, a few a
300 /// day, in case an event never came.
301 pub(crate) async fn refresh_from_stripe(&self) -> Result<String> {
302 if self.stripe.is_none() {
303 return Ok("payments are not set up".to_owned());
304 }
305 #[derive(Deserialize)]
306 struct Customer {
307 customer_id: String,
308 }
309 #[derive(Deserialize)]
310 struct Plan {
311 subscription_id: String,
312 }
313 let card_cutoff = rfc3339(now_ms().saturating_sub(CARD_FRESH_DAYS * 24 * 60 * 60 * 1000));
314 let customers = self
315 .db
316 .prepare(
317 "SELECT DISTINCT customer_id FROM accounts
318 WHERE customer_id IS NOT NULL AND card_synced_at IS NOT NULL AND card_synced_at < ?1
319 ORDER BY card_synced_at LIMIT ?2",
320 )
321 .bind(&[card_cutoff.into(), REFRESH_BATCH.into()])?
322 .all()
323 .await?
324 .results::<Customer>()?;
325 let plan_cutoff = rfc3339(now_ms().saturating_sub(24 * 60 * 60 * 1000));
326 let plans = self
327 .db
328 .prepare(
329 "SELECT subscription_id FROM subscriptions WHERE status <> 'canceled' AND updated_at < ?1
330 ORDER BY updated_at LIMIT ?2",
331 )
332 .bind(&[plan_cutoff.into(), REFRESH_BATCH.into()])?
333 .all()
334 .await?
335 .results::<Plan>()?;
336 let mut failed = 0;
337 for customer in &customers {
338 if let Err(error) = self.fetch_card(&customer.customer_id).await {
339 worker::console_error!("refreshing {}'s card failed: {error}", customer.customer_id);
340 failed += 1;
341 }
342 }
343 for plan in &plans {
344 if let Err(error) = self.settle_subscription(&plan.subscription_id).await {
345 worker::console_error!("refreshing plan {} failed: {error}", plan.subscription_id);
346 failed += 1;
347 }
348 }
349 Ok(format!("{} cards and {} plans read again, {failed} failed", customers.len(), plans.len()))
350 }
351}
352
353fn card_of(row: &CardRow) -> Option<Card> {
354 Some(Card {
355 brand: row.card_brand.clone()?,
356 last4: row.card_last4.clone()?,
357 exp_month: row.card_exp_month?,
358 exp_year: row.card_exp_year?,
359 })
360}
361
362#[cfg(test)]
363mod tests {
364
365 #[test]
366 fn no_binding_turns_none_into_undefined() {
367 // D1 refuses `undefined`; every optional value goes through or_null.
368 let source = include_str!("stripe_sync.rs");
369 let body = source.split("#[cfg(test)]").next().unwrap();
370 for line in body.lines().filter(|l| l.trim_end().ends_with(".into(),")) {
371 assert!(!line.contains("then_some(") && !line.contains(".map(|"), "{line}");
372 }
373 }
374
375 use super::*;
376
377 #[test]
378 fn the_replay_moves_on_only_past_what_it_handled() {
379 let now = 1_791_000_000;
380 // All handled: just behind the clock.
381 assert_eq!(next_through(now - 900, now, None, true), now - SETTLE_SECONDS);
382 // Never backwards.
383 assert_eq!(next_through(now, now, None, true), now);
384 // A failure: back to it, so it is listed again.
385 assert_eq!(next_through(now - 900, now, Some(now - 600), true), now - 600);
386 assert_eq!(next_through(now - 900, now, Some(now - 1200), true), now - 1200);
387 // The list not read to its end: stay.
388 assert_eq!(next_through(now - 900, now, None, false), now - 900);
389 // A failed event on a partial page must not skip older unread events.
390 assert_eq!(next_through(now - 900, now, Some(now - 100), false), now - 900);
391 }
392
393 #[test]
394 fn a_card_needs_every_part() {
395 let row = |brand: Option<&str>| CardRow {
396 customer_id: Some("cus_1".into()),
397 card_brand: brand.map(str::to_owned),
398 card_last4: Some("4242".into()),
399 card_exp_month: Some(12),
400 card_exp_year: Some(2030),
401 card_synced_at: Some("2026-10-06T00:00:00Z".into()),
402 };
403 assert_eq!(card_of(&row(Some("visa"))).map(|c| c.last4), Some("4242".into()));
404 assert!(card_of(&row(None)).is_none());
405 }
406}