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
370 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) => created.min(previous.max(created)),
61 None if !complete => previous,
62 None => previous.max(now - SETTLE_SECONDS),
63 }
64}
65
66impl Billing {
67 /// The workspace's saved card as last synced. Stripe is asked only when
68 /// it never was, or g1t changed the card since; a Stripe that does not
69 /// answer then shows no card, as before.
70 pub(crate) async fn saved_card(&self, workspace: &str) -> Result<Option<Card>> {
71 let Some(row) = self
72 .db
73 .prepare(
74 "SELECT customer_id, card_brand, card_last4, card_exp_month, card_exp_year, card_synced_at
75 FROM accounts WHERE workspace = ?",
76 )
77 .bind(&[workspace.into()])?
78 .first::<CardRow>(None)
79 .await?
80 else {
81 return Ok(None);
82 };
83 if row.card_synced_at.is_some() {
84 return Ok(card_of(&row));
85 }
86 let Some(customer) = row.customer_id.as_deref() else { return Ok(None) };
87 if self.stripe.is_none() {
88 return Ok(None);
89 }
90 Ok(match self.fetch_card(customer).await {
91 Ok(card) => card,
92 Err(error) => {
93 worker::console_error!("reading {workspace}'s card from Stripe failed: {error}");
94 None
95 }
96 })
97 }
98
99 /// Asks Stripe for the customer's card and keeps it on every account
100 /// with that customer.
101 async fn fetch_card(&self, customer: &str) -> Result<Option<Card>> {
102 let Some(stripe) = &self.stripe else { return Ok(None) };
103 // The card invoices are charged to (the default), not merely the
104 // newest: what Stripe's billing page made the default is the one.
105 let method = match stripe.default_payment_method(customer).await {
106 Ok(method) => method,
107 // Deleted at Stripe, or from another account: every account
108 // with it forgets it, so the next card saved makes a new one.
109 Err(error) if crate::stripe::is_missing(&error) => {
110 self.db
111 .prepare(
112 "UPDATE accounts SET customer_id = NULL, card_brand = NULL, card_last4 = NULL, card_exp_month = NULL,
113 card_exp_year = NULL, card_synced_at = NULL WHERE customer_id = ?",
114 )
115 .bind(&[customer.into()])?
116 .run()
117 .await?;
118 return Ok(None);
119 }
120 Err(error) => return Err(error),
121 };
122 let card = method.filter(|m| m.kind == "card").and_then(|m| {
123 Some(Card { brand: m.brand?, last4: m.last4?, exp_month: m.exp_month?, exp_year: m.exp_year? })
124 });
125 self.db
126 .prepare(
127 "UPDATE accounts SET card_brand = ?2, card_last4 = ?3, card_exp_month = ?4, card_exp_year = ?5,
128 card_synced_at = ?6 WHERE customer_id = ?1",
129 )
130 .bind(&[
131 customer.into(),
132 card.as_ref().map(|c| c.brand.clone()).into(),
133 card.as_ref().map(|c| c.last4.clone()).into(),
134 card.as_ref().map(|c| c.exp_month as f64).into(),
135 card.as_ref().map(|c| c.exp_year as f64).into(),
136 rfc3339(now_ms()).into(),
137 ])?
138 .run()
139 .await?;
140 Ok(card)
141 }
142
143 /// A card or customer event: the saved card read again. The event's
144 /// outcome, for `stripe_events`.
145 pub(crate) async fn sync_card_of(&self, customer: &str) -> Result<String> {
146 if self.workspace_of_customer(customer).await?.is_none() {
147 return Ok("ignored: not a workspace's customer".to_owned());
148 }
149 Ok(match self.fetch_card(customer).await? {
150 Some(card) => format!("card saved: {} ending {}", card.brand, card.last4),
151 None => "card saved: none".to_owned(),
152 })
153 }
154
155 /// Forgets the saved card after g1t itself changed it, so the next
156 /// read asks Stripe instead of waiting for the event.
157 pub(crate) async fn forget_card(&self, workspace: &str) -> Result<()> {
158 self.db
159 .prepare("UPDATE accounts SET card_synced_at = NULL WHERE workspace = ?")
160 .bind(&[workspace.into()])?
161 .run()
162 .await?;
163 Ok(())
164 }
165
166 /// Handles events Stripe sent that billing never saw, from its event
167 /// list. What happened, for the log.
168 pub(crate) async fn replay_events(&self) -> Result<String> {
169 let Some(stripe) = &self.stripe else { return Ok("payments are not set up".to_owned()) };
170 #[derive(Deserialize)]
171 struct Mark {
172 through: i64,
173 }
174 #[derive(Deserialize)]
175 struct Page {
176 data: Vec<Value>,
177 has_more: bool,
178 }
179 let mode = self.mode();
180 let now = (now_ms() / 1000) as i64;
181 // A claim whose handler died never finished: let it be handled again.
182 self.db
183 .prepare("DELETE FROM stripe_events WHERE outcome = 'handling' AND received_at < ?")
184 .bind(&[rfc3339(now_ms().saturating_sub(STALE_CLAIM_MS)).into()])?
185 .run()
186 .await?;
187 let previous = self
188 .db
189 .prepare("SELECT through FROM stripe_sync WHERE mode = ?")
190 .bind(&[mode.into()])?
191 .first::<Mark>(None)
192 .await?
193 .map_or(now - FIRST_LOOK_SECONDS, |mark| mark.through);
194 let since = previous - OVERLAP_SECONDS;
195
196 // Stripe lists newest first; read back to `since`.
197 let mut listed: Vec<Value> = Vec::new();
198 let mut complete = false;
199 let mut after: Option<String> = None;
200 for _ in 0..MAX_PAGES {
201 let mut query: Vec<(&str, String)> = vec![("limit", "100".to_owned()), ("created[gte]", since.to_string())];
202 query.extend(EVENTS.iter().map(|event| ("types[]", (*event).to_owned())));
203 if let Some(after) = &after {
204 query.push(("starting_after", after.clone()));
205 }
206 let page: Page = stripe.get(&format!("/events?{}", form(&query))).await?;
207 after = page.data.last().and_then(|event| event["id"].as_str()).map(str::to_owned);
208 listed.extend(page.data);
209 if !page.has_more || after.is_none() {
210 complete = true;
211 break;
212 }
213 }
214 if !complete {
215 worker::console_error!("Stripe listed more than {} events since {since}; replaying the newest", MAX_PAGES * 100);
216 }
217
218 // Oldest first, as they happened.
219 listed.sort_by_key(|event| event["created"].as_i64().unwrap_or(0));
220 let mut handled = 0;
221 let mut first_failed = None;
222 for event in &listed {
223 match self.process_event(event).await {
224 Ok(true) => handled += 1,
225 Ok(false) => {}
226 Err(error) => {
227 let id = event["id"].as_str().unwrap_or_default();
228 worker::console_error!("replaying Stripe event {id} failed: {error}");
229 first_failed = first_failed.or(event["created"].as_i64());
230 }
231 }
232 }
233 let through = next_through(previous, now, first_failed, complete);
234 self.db
235 .prepare(
236 "INSERT INTO stripe_sync (mode, through, checked_at) VALUES (?1, ?2, ?3)
237 ON CONFLICT (mode) DO UPDATE SET through = ?2, checked_at = ?3",
238 )
239 .bind(&[mode.into(), (through as f64).into(), rfc3339(now_ms()).into()])?
240 .run()
241 .await?;
242 Ok(format!("{} listed, {handled} not seen before and handled", listed.len()))
243 }
244
245 /// Keeps the destination at billing's address enabled and sending
246 /// every event billing handles. Its signing secret is not touched.
247 /// Nothing when Stripe has no destination there.
248 pub(crate) async fn keep_endpoint(&self, by: &str) -> Result<String> {
249 let (Some(stripe), Some(destination)) = (&self.stripe, self.destination().await?) else {
250 return Ok(format!("no destination at {WEBHOOK_URL}"));
251 };
252 let path = format!("/webhook_endpoints/{}", destination.id);
253 let mut fields: Vec<(String, String)> = Vec::new();
254 let mut changes = Vec::new();
255 if destination.status != "enabled" {
256 fields.push(("disabled".to_owned(), "false".to_owned()));
257 changes.push("enabled again".to_owned());
258 }
259 let missing = missing_events(&destination.enabled_events);
260 if !missing.is_empty() {
261 // Stripe replaces the list: what it sends now, plus what is missing.
262 let all = destination.enabled_events.iter().cloned().chain(missing.iter().cloned());
263 fields.extend(all.enumerate().map(|(i, event)| (format!("enabled_events[{i}]"), event)));
264 changes.push(format!("added {}", missing.join(", ")));
265 }
266 if changes.is_empty() {
267 return Ok("destination as it should be".to_owned());
268 }
269 let fields: Vec<(&str, String)> = fields.iter().map(|(name, value)| (name.as_str(), value.clone())).collect();
270 let _: Value = stripe.post(&path, &fields).await?;
271 let done = changes.join("; ");
272 self.audit("stripe", "webhook", &format!("Destination {}: {done}", destination.id), by).await?;
273 Ok(done)
274 }
275
276 /// Reads again the saved cards and plans not read in a while, a few a
277 /// day, in case an event never came.
278 pub(crate) async fn refresh_from_stripe(&self) -> Result<String> {
279 if self.stripe.is_none() {
280 return Ok("payments are not set up".to_owned());
281 }
282 #[derive(Deserialize)]
283 struct Customer {
284 customer_id: String,
285 }
286 #[derive(Deserialize)]
287 struct Plan {
288 subscription_id: String,
289 }
290 let card_cutoff = rfc3339(now_ms().saturating_sub(CARD_FRESH_DAYS * 24 * 60 * 60 * 1000));
291 let customers = self
292 .db
293 .prepare(
294 "SELECT DISTINCT customer_id FROM accounts
295 WHERE customer_id IS NOT NULL AND card_synced_at IS NOT NULL AND card_synced_at < ?1
296 ORDER BY card_synced_at LIMIT ?2",
297 )
298 .bind(&[card_cutoff.into(), REFRESH_BATCH.into()])?
299 .all()
300 .await?
301 .results::<Customer>()?;
302 let plan_cutoff = rfc3339(now_ms().saturating_sub(24 * 60 * 60 * 1000));
303 let plans = self
304 .db
305 .prepare(
306 "SELECT subscription_id FROM subscriptions WHERE status <> 'canceled' AND updated_at < ?1
307 ORDER BY updated_at LIMIT ?2",
308 )
309 .bind(&[plan_cutoff.into(), REFRESH_BATCH.into()])?
310 .all()
311 .await?
312 .results::<Plan>()?;
313 let mut failed = 0;
314 for customer in &customers {
315 if let Err(error) = self.fetch_card(&customer.customer_id).await {
316 worker::console_error!("refreshing {}'s card failed: {error}", customer.customer_id);
317 failed += 1;
318 }
319 }
320 for plan in &plans {
321 if let Err(error) = self.settle_subscription(&plan.subscription_id).await {
322 worker::console_error!("refreshing plan {} failed: {error}", plan.subscription_id);
323 failed += 1;
324 }
325 }
326 Ok(format!("{} cards and {} plans read again, {failed} failed", customers.len(), plans.len()))
327 }
328}
329
330fn card_of(row: &CardRow) -> Option<Card> {
331 Some(Card {
332 brand: row.card_brand.clone()?,
333 last4: row.card_last4.clone()?,
334 exp_month: row.card_exp_month?,
335 exp_year: row.card_exp_year?,
336 })
337}
338
339#[cfg(test)]
340mod tests {
341 use super::*;
342
343 #[test]
344 fn the_replay_moves_on_only_past_what_it_handled() {
345 let now = 1_791_000_000;
346 // All handled: just behind the clock.
347 assert_eq!(next_through(now - 900, now, None, true), now - SETTLE_SECONDS);
348 // Never backwards.
349 assert_eq!(next_through(now, now, None, true), now);
350 // A failure: back to it, so it is listed again.
351 assert_eq!(next_through(now - 900, now, Some(now - 600), true), now - 600);
352 assert_eq!(next_through(now - 900, now, Some(now - 1200), true), now - 1200);
353 // The list not read to its end: stay.
354 assert_eq!(next_through(now - 900, now, None, false), now - 900);
355 }
356
357 #[test]
358 fn a_card_needs_every_part() {
359 let row = |brand: Option<&str>| CardRow {
360 customer_id: Some("cus_1".into()),
361 card_brand: brand.map(str::to_owned),
362 card_last4: Some("4242".into()),
363 card_exp_month: Some(12),
364 card_exp_year: Some(2030),
365 card_synced_at: Some("2026-10-06T00:00:00Z".into()),
366 };
367 assert_eq!(card_of(&row(Some("visa"))).map(|c| c.last4), Some("4242".into()));
368 assert!(card_of(&row(None)).is_none());
369 }
370}