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