1use std::str::FromStr;
2use std::sync::Arc;
3use std::time::Duration;
4
5use anyhow::{anyhow, bail, Context};
6use axum::{
7 extract::{self, Query},
8 routing::{get, post},
9 Extension, Json, Router,
10};
11use chrono::{DateTime, SecondsFormat, Utc};
12use reqwest::StatusCode;
13use sea_orm::ActiveValue;
14use serde::{Deserialize, Serialize};
15use stripe::{
16 BillingPortalSession, CheckoutSession, CreateBillingPortalSession,
17 CreateBillingPortalSessionFlowData, CreateBillingPortalSessionFlowDataAfterCompletion,
18 CreateBillingPortalSessionFlowDataAfterCompletionRedirect,
19 CreateBillingPortalSessionFlowDataType, CreateCheckoutSession, CreateCheckoutSessionLineItems,
20 CreateCustomer, Customer, CustomerId, EventObject, EventType, Expandable, ListEvents,
21 Subscription, SubscriptionId, SubscriptionStatus,
22};
23use util::ResultExt;
24
25use crate::db::billing_subscription::StripeSubscriptionStatus;
26use crate::db::{
27 billing_customer, BillingSubscriptionId, CreateBillingCustomerParams,
28 CreateBillingSubscriptionParams, CreateProcessedStripeEventParams, UpdateBillingCustomerParams,
29 UpdateBillingSubscriptionParams,
30};
31use crate::{AppState, Error, Result};
32
33pub fn router() -> Router {
34 Router::new()
35 .route(
36 "/billing/subscriptions",
37 get(list_billing_subscriptions).post(create_billing_subscription),
38 )
39 .route(
40 "/billing/subscriptions/manage",
41 post(manage_billing_subscription),
42 )
43}
44
45#[derive(Debug, Deserialize)]
46struct ListBillingSubscriptionsParams {
47 github_user_id: i32,
48}
49
50#[derive(Debug, Serialize)]
51struct BillingSubscriptionJson {
52 id: BillingSubscriptionId,
53 name: String,
54 status: StripeSubscriptionStatus,
55 cancel_at: Option<String>,
56 /// Whether this subscription can be canceled.
57 is_cancelable: bool,
58}
59
60#[derive(Debug, Serialize)]
61struct ListBillingSubscriptionsResponse {
62 subscriptions: Vec<BillingSubscriptionJson>,
63}
64
65async fn list_billing_subscriptions(
66 Extension(app): Extension<Arc<AppState>>,
67 Query(params): Query<ListBillingSubscriptionsParams>,
68) -> Result<Json<ListBillingSubscriptionsResponse>> {
69 let user = app
70 .db
71 .get_user_by_github_user_id(params.github_user_id)
72 .await?
73 .ok_or_else(|| anyhow!("user not found"))?;
74
75 let subscriptions = app.db.get_billing_subscriptions(user.id).await?;
76
77 Ok(Json(ListBillingSubscriptionsResponse {
78 subscriptions: subscriptions
79 .into_iter()
80 .map(|subscription| BillingSubscriptionJson {
81 id: subscription.id,
82 name: "Zed Pro".to_string(),
83 status: subscription.stripe_subscription_status,
84 cancel_at: subscription.stripe_cancel_at.map(|cancel_at| {
85 cancel_at
86 .and_utc()
87 .to_rfc3339_opts(SecondsFormat::Millis, true)
88 }),
89 is_cancelable: subscription.stripe_subscription_status.is_cancelable()
90 && subscription.stripe_cancel_at.is_none(),
91 })
92 .collect(),
93 }))
94}
95
96#[derive(Debug, Deserialize)]
97struct CreateBillingSubscriptionBody {
98 github_user_id: i32,
99}
100
101#[derive(Debug, Serialize)]
102struct CreateBillingSubscriptionResponse {
103 checkout_session_url: String,
104}
105
106/// Initiates a Stripe Checkout session for creating a billing subscription.
107async fn create_billing_subscription(
108 Extension(app): Extension<Arc<AppState>>,
109 extract::Json(body): extract::Json<CreateBillingSubscriptionBody>,
110) -> Result<Json<CreateBillingSubscriptionResponse>> {
111 let user = app
112 .db
113 .get_user_by_github_user_id(body.github_user_id)
114 .await?
115 .ok_or_else(|| anyhow!("user not found"))?;
116
117 let Some((stripe_client, stripe_price_id)) = app
118 .stripe_client
119 .clone()
120 .zip(app.config.stripe_price_id.clone())
121 else {
122 log::error!("failed to retrieve Stripe client or price ID");
123 Err(Error::http(
124 StatusCode::NOT_IMPLEMENTED,
125 "not supported".into(),
126 ))?
127 };
128
129 let customer_id =
130 if let Some(existing_customer) = app.db.get_billing_customer_by_user_id(user.id).await? {
131 CustomerId::from_str(&existing_customer.stripe_customer_id)
132 .context("failed to parse customer ID")?
133 } else {
134 let customer = Customer::create(
135 &stripe_client,
136 CreateCustomer {
137 email: user.email_address.as_deref(),
138 ..Default::default()
139 },
140 )
141 .await?;
142
143 customer.id
144 };
145
146 let checkout_session = {
147 let mut params = CreateCheckoutSession::new();
148 params.mode = Some(stripe::CheckoutSessionMode::Subscription);
149 params.customer = Some(customer_id);
150 params.client_reference_id = Some(user.github_login.as_str());
151 params.line_items = Some(vec![CreateCheckoutSessionLineItems {
152 price: Some(stripe_price_id.to_string()),
153 quantity: Some(1),
154 ..Default::default()
155 }]);
156 let success_url = format!("{}/account", app.config.zed_dot_dev_url());
157 params.success_url = Some(&success_url);
158
159 CheckoutSession::create(&stripe_client, params).await?
160 };
161
162 Ok(Json(CreateBillingSubscriptionResponse {
163 checkout_session_url: checkout_session
164 .url
165 .ok_or_else(|| anyhow!("no checkout session URL"))?,
166 }))
167}
168
169#[derive(Debug, PartialEq, Deserialize)]
170#[serde(rename_all = "snake_case")]
171enum ManageSubscriptionIntent {
172 /// The user intends to cancel their subscription.
173 Cancel,
174 /// The user intends to stop the cancellation of their subscription.
175 StopCancellation,
176}
177
178#[derive(Debug, Deserialize)]
179struct ManageBillingSubscriptionBody {
180 github_user_id: i32,
181 intent: ManageSubscriptionIntent,
182 /// The ID of the subscription to manage.
183 subscription_id: BillingSubscriptionId,
184}
185
186#[derive(Debug, Serialize)]
187struct ManageBillingSubscriptionResponse {
188 billing_portal_session_url: Option<String>,
189}
190
191/// Initiates a Stripe customer portal session for managing a billing subscription.
192async fn manage_billing_subscription(
193 Extension(app): Extension<Arc<AppState>>,
194 extract::Json(body): extract::Json<ManageBillingSubscriptionBody>,
195) -> Result<Json<ManageBillingSubscriptionResponse>> {
196 let user = app
197 .db
198 .get_user_by_github_user_id(body.github_user_id)
199 .await?
200 .ok_or_else(|| anyhow!("user not found"))?;
201
202 let Some(stripe_client) = app.stripe_client.clone() else {
203 log::error!("failed to retrieve Stripe client");
204 Err(Error::http(
205 StatusCode::NOT_IMPLEMENTED,
206 "not supported".into(),
207 ))?
208 };
209
210 let customer = app
211 .db
212 .get_billing_customer_by_user_id(user.id)
213 .await?
214 .ok_or_else(|| anyhow!("billing customer not found"))?;
215 let customer_id = CustomerId::from_str(&customer.stripe_customer_id)
216 .context("failed to parse customer ID")?;
217
218 let subscription = app
219 .db
220 .get_billing_subscription_by_id(body.subscription_id)
221 .await?
222 .ok_or_else(|| anyhow!("subscription not found"))?;
223
224 if body.intent == ManageSubscriptionIntent::StopCancellation {
225 let subscription_id = SubscriptionId::from_str(&subscription.stripe_subscription_id)
226 .context("failed to parse subscription ID")?;
227
228 let updated_stripe_subscription = Subscription::update(
229 &stripe_client,
230 &subscription_id,
231 stripe::UpdateSubscription {
232 cancel_at_period_end: Some(false),
233 ..Default::default()
234 },
235 )
236 .await?;
237
238 app.db
239 .update_billing_subscription(
240 subscription.id,
241 &UpdateBillingSubscriptionParams {
242 stripe_cancel_at: ActiveValue::set(
243 updated_stripe_subscription
244 .cancel_at
245 .and_then(|cancel_at| DateTime::from_timestamp(cancel_at, 0))
246 .map(|time| time.naive_utc()),
247 ),
248 ..Default::default()
249 },
250 )
251 .await?;
252
253 return Ok(Json(ManageBillingSubscriptionResponse {
254 billing_portal_session_url: None,
255 }));
256 }
257
258 let flow = match body.intent {
259 ManageSubscriptionIntent::Cancel => CreateBillingPortalSessionFlowData {
260 type_: CreateBillingPortalSessionFlowDataType::SubscriptionCancel,
261 after_completion: Some(CreateBillingPortalSessionFlowDataAfterCompletion {
262 type_: stripe::CreateBillingPortalSessionFlowDataAfterCompletionType::Redirect,
263 redirect: Some(CreateBillingPortalSessionFlowDataAfterCompletionRedirect {
264 return_url: format!("{}/account", app.config.zed_dot_dev_url()),
265 }),
266 ..Default::default()
267 }),
268 subscription_cancel: Some(
269 stripe::CreateBillingPortalSessionFlowDataSubscriptionCancel {
270 subscription: subscription.stripe_subscription_id,
271 retention: None,
272 },
273 ),
274 ..Default::default()
275 },
276 ManageSubscriptionIntent::StopCancellation => unreachable!(),
277 };
278
279 let mut params = CreateBillingPortalSession::new(customer_id);
280 params.flow_data = Some(flow);
281 let return_url = format!("{}/account", app.config.zed_dot_dev_url());
282 params.return_url = Some(&return_url);
283
284 let session = BillingPortalSession::create(&stripe_client, params).await?;
285
286 Ok(Json(ManageBillingSubscriptionResponse {
287 billing_portal_session_url: Some(session.url),
288 }))
289}
290
291/// The amount of time we wait in between each poll of Stripe events.
292///
293/// This value should strike a balance between:
294/// 1. Being short enough that we update quickly when something in Stripe changes
295/// 2. Being long enough that we don't eat into our rate limits.
296///
297/// As a point of reference, the Sequin folks say they have this at **500ms**:
298///
299/// > We poll the Stripe /events endpoint every 500ms per account
300/// >
301/// > — https://blog.sequinstream.com/events-not-webhooks/
302const POLL_EVENTS_INTERVAL: Duration = Duration::from_secs(5);
303
304/// The maximum number of events to return per page.
305///
306/// We set this to 100 (the max) so we have to make fewer requests to Stripe.
307///
308/// > Limit can range between 1 and 100, and the default is 10.
309const EVENTS_LIMIT_PER_PAGE: u64 = 100;
310
311/// The number of pages consisting entirely of already-processed events that we
312/// will see before we stop retrieving events.
313///
314/// This is used to prevent over-fetching the Stripe events API for events we've
315/// already seen and processed.
316const NUMBER_OF_ALREADY_PROCESSED_PAGES_BEFORE_WE_STOP: usize = 4;
317
318/// Polls the Stripe events API periodically to reconcile the records in our
319/// database with the data in Stripe.
320pub fn poll_stripe_events_periodically(app: Arc<AppState>) {
321 let Some(stripe_client) = app.stripe_client.clone() else {
322 log::warn!("failed to retrieve Stripe client");
323 return;
324 };
325
326 let executor = app.executor.clone();
327 executor.spawn_detached({
328 let executor = executor.clone();
329 async move {
330 loop {
331 poll_stripe_events(&app, &stripe_client).await.log_err();
332
333 executor.sleep(POLL_EVENTS_INTERVAL).await;
334 }
335 }
336 });
337}
338
339async fn poll_stripe_events(
340 app: &Arc<AppState>,
341 stripe_client: &stripe::Client,
342) -> anyhow::Result<()> {
343 fn event_type_to_string(event_type: EventType) -> String {
344 // Calling `to_string` on `stripe::EventType` members gives us a quoted string,
345 // so we need to unquote it.
346 event_type.to_string().trim_matches('"').to_string()
347 }
348
349 let event_types = [
350 EventType::CustomerCreated,
351 EventType::CustomerUpdated,
352 EventType::CustomerSubscriptionCreated,
353 EventType::CustomerSubscriptionUpdated,
354 EventType::CustomerSubscriptionPaused,
355 EventType::CustomerSubscriptionResumed,
356 EventType::CustomerSubscriptionDeleted,
357 ]
358 .into_iter()
359 .map(event_type_to_string)
360 .collect::<Vec<_>>();
361
362 let mut pages_of_already_processed_events = 0;
363 let mut unprocessed_events = Vec::new();
364
365 loop {
366 if pages_of_already_processed_events >= NUMBER_OF_ALREADY_PROCESSED_PAGES_BEFORE_WE_STOP {
367 log::info!("saw {pages_of_already_processed_events} pages of already-processed events: stopping event retrieval");
368 break;
369 }
370
371 log::info!("retrieving events from Stripe: {}", event_types.join(", "));
372
373 let mut params = ListEvents::new();
374 params.types = Some(event_types.clone());
375 params.limit = Some(EVENTS_LIMIT_PER_PAGE);
376
377 let events = stripe::Event::list(stripe_client, ¶ms).await?;
378
379 let processed_event_ids = {
380 let event_ids = &events
381 .data
382 .iter()
383 .map(|event| event.id.as_str())
384 .collect::<Vec<_>>();
385
386 app.db
387 .get_processed_stripe_events_by_event_ids(event_ids)
388 .await?
389 .into_iter()
390 .map(|event| event.stripe_event_id)
391 .collect::<Vec<_>>()
392 };
393
394 let mut processed_events_in_page = 0;
395 let events_in_page = events.data.len();
396 for event in events.data {
397 if processed_event_ids.contains(&event.id.to_string()) {
398 processed_events_in_page += 1;
399 log::debug!("Stripe event {} already processed: skipping", event.id);
400 } else {
401 unprocessed_events.push(event);
402 }
403 }
404
405 if processed_events_in_page == events_in_page {
406 pages_of_already_processed_events += 1;
407 }
408
409 if !events.has_more {
410 break;
411 }
412 }
413
414 log::info!(
415 "unprocessed events from Stripe: {}",
416 unprocessed_events.len()
417 );
418
419 // Sort all of the unprocessed events in ascending order, so we can handle them in the order they occurred.
420 unprocessed_events.sort_by(|a, b| a.created.cmp(&b.created).then_with(|| a.id.cmp(&b.id)));
421
422 for event in unprocessed_events {
423 let event_id = event.id.clone();
424 let processed_event_params = CreateProcessedStripeEventParams {
425 stripe_event_id: event.id.to_string(),
426 stripe_event_type: event_type_to_string(event.type_),
427 stripe_event_created_timestamp: event.created,
428 };
429
430 // If the event has happened too far in the past, we don't want to
431 // process it and risk overwriting other more-recent updates.
432 //
433 // 1 hour was chosen arbitrarily. This could be made longer or shorter.
434 let one_hour = Duration::from_secs(60 * 60);
435 let an_hour_ago = Utc::now() - one_hour;
436 if an_hour_ago.timestamp() > event.created {
437 log::info!(
438 "Stripe event {} is more than {one_hour:?} old, marking as processed",
439 event_id
440 );
441 app.db
442 .create_processed_stripe_event(&processed_event_params)
443 .await?;
444
445 return Ok(());
446 }
447
448 let process_result = match event.type_ {
449 EventType::CustomerCreated | EventType::CustomerUpdated => {
450 handle_customer_event(app, stripe_client, event).await
451 }
452 EventType::CustomerSubscriptionCreated
453 | EventType::CustomerSubscriptionUpdated
454 | EventType::CustomerSubscriptionPaused
455 | EventType::CustomerSubscriptionResumed
456 | EventType::CustomerSubscriptionDeleted => {
457 handle_customer_subscription_event(app, stripe_client, event).await
458 }
459 _ => Ok(()),
460 };
461
462 if let Some(()) = process_result
463 .with_context(|| format!("failed to process event {event_id} successfully"))
464 .log_err()
465 {
466 app.db
467 .create_processed_stripe_event(&processed_event_params)
468 .await?;
469 }
470 }
471
472 Ok(())
473}
474
475async fn handle_customer_event(
476 app: &Arc<AppState>,
477 _stripe_client: &stripe::Client,
478 event: stripe::Event,
479) -> anyhow::Result<()> {
480 let EventObject::Customer(customer) = event.data.object else {
481 bail!("unexpected event payload for {}", event.id);
482 };
483
484 log::info!("handling Stripe {} event: {}", event.type_, event.id);
485
486 let Some(email) = customer.email else {
487 log::info!("Stripe customer has no email: skipping");
488 return Ok(());
489 };
490
491 let Some(user) = app.db.get_user_by_email(&email).await? else {
492 log::info!("no user found for email: skipping");
493 return Ok(());
494 };
495
496 if let Some(existing_customer) = app
497 .db
498 .get_billing_customer_by_stripe_customer_id(&customer.id)
499 .await?
500 {
501 app.db
502 .update_billing_customer(
503 existing_customer.id,
504 &UpdateBillingCustomerParams {
505 // For now we just leave the information as-is, as it is not
506 // likely to change.
507 ..Default::default()
508 },
509 )
510 .await?;
511 } else {
512 app.db
513 .create_billing_customer(&CreateBillingCustomerParams {
514 user_id: user.id,
515 stripe_customer_id: customer.id.to_string(),
516 })
517 .await?;
518 }
519
520 Ok(())
521}
522
523async fn handle_customer_subscription_event(
524 app: &Arc<AppState>,
525 stripe_client: &stripe::Client,
526 event: stripe::Event,
527) -> anyhow::Result<()> {
528 let EventObject::Subscription(subscription) = event.data.object else {
529 bail!("unexpected event payload for {}", event.id);
530 };
531
532 log::info!("handling Stripe {} event: {}", event.type_, event.id);
533
534 let billing_customer =
535 find_or_create_billing_customer(app, stripe_client, subscription.customer)
536 .await?
537 .ok_or_else(|| anyhow!("billing customer not found"))?;
538
539 if let Some(existing_subscription) = app
540 .db
541 .get_billing_subscription_by_stripe_subscription_id(&subscription.id)
542 .await?
543 {
544 app.db
545 .update_billing_subscription(
546 existing_subscription.id,
547 &UpdateBillingSubscriptionParams {
548 billing_customer_id: ActiveValue::set(billing_customer.id),
549 stripe_subscription_id: ActiveValue::set(subscription.id.to_string()),
550 stripe_subscription_status: ActiveValue::set(subscription.status.into()),
551 stripe_cancel_at: ActiveValue::set(
552 subscription
553 .cancel_at
554 .and_then(|cancel_at| DateTime::from_timestamp(cancel_at, 0))
555 .map(|time| time.naive_utc()),
556 ),
557 },
558 )
559 .await?;
560 } else {
561 app.db
562 .create_billing_subscription(&CreateBillingSubscriptionParams {
563 billing_customer_id: billing_customer.id,
564 stripe_subscription_id: subscription.id.to_string(),
565 stripe_subscription_status: subscription.status.into(),
566 })
567 .await?;
568 }
569
570 Ok(())
571}
572
573impl From<SubscriptionStatus> for StripeSubscriptionStatus {
574 fn from(value: SubscriptionStatus) -> Self {
575 match value {
576 SubscriptionStatus::Incomplete => Self::Incomplete,
577 SubscriptionStatus::IncompleteExpired => Self::IncompleteExpired,
578 SubscriptionStatus::Trialing => Self::Trialing,
579 SubscriptionStatus::Active => Self::Active,
580 SubscriptionStatus::PastDue => Self::PastDue,
581 SubscriptionStatus::Canceled => Self::Canceled,
582 SubscriptionStatus::Unpaid => Self::Unpaid,
583 SubscriptionStatus::Paused => Self::Paused,
584 }
585 }
586}
587
588/// Finds or creates a billing customer using the provided customer.
589async fn find_or_create_billing_customer(
590 app: &Arc<AppState>,
591 stripe_client: &stripe::Client,
592 customer_or_id: Expandable<Customer>,
593) -> anyhow::Result<Option<billing_customer::Model>> {
594 let customer_id = match &customer_or_id {
595 Expandable::Id(id) => id,
596 Expandable::Object(customer) => customer.id.as_ref(),
597 };
598
599 // If we already have a billing customer record associated with the Stripe customer,
600 // there's nothing more we need to do.
601 if let Some(billing_customer) = app
602 .db
603 .get_billing_customer_by_stripe_customer_id(&customer_id)
604 .await?
605 {
606 return Ok(Some(billing_customer));
607 }
608
609 // If all we have is a customer ID, resolve it to a full customer record by
610 // hitting the Stripe API.
611 let customer = match customer_or_id {
612 Expandable::Id(id) => Customer::retrieve(&stripe_client, &id, &[]).await?,
613 Expandable::Object(customer) => *customer,
614 };
615
616 let Some(email) = customer.email else {
617 return Ok(None);
618 };
619
620 let Some(user) = app.db.get_user_by_email(&email).await? else {
621 return Ok(None);
622 };
623
624 let billing_customer = app
625 .db
626 .create_billing_customer(&CreateBillingCustomerParams {
627 user_id: user.id,
628 stripe_customer_id: customer.id.to_string(),
629 })
630 .await?;
631
632 Ok(Some(billing_customer))
633}