1use anyhow::{Context, anyhow, bail};
2use axum::{
3 Extension, Json, Router,
4 extract::{self, Query},
5 routing::{get, post},
6};
7use chrono::{DateTime, SecondsFormat, Utc};
8use collections::HashSet;
9use reqwest::StatusCode;
10use sea_orm::ActiveValue;
11use serde::{Deserialize, Serialize};
12use serde_json::json;
13use std::{str::FromStr, sync::Arc, time::Duration};
14use stripe::{
15 BillingPortalSession, CancellationDetailsReason, CreateBillingPortalSession,
16 CreateBillingPortalSessionFlowData, CreateBillingPortalSessionFlowDataAfterCompletion,
17 CreateBillingPortalSessionFlowDataAfterCompletionRedirect,
18 CreateBillingPortalSessionFlowDataSubscriptionUpdateConfirm,
19 CreateBillingPortalSessionFlowDataSubscriptionUpdateConfirmItems,
20 CreateBillingPortalSessionFlowDataType, CreateCustomer, Customer, CustomerId, EventObject,
21 EventType, Expandable, ListEvents, Subscription, SubscriptionId, SubscriptionStatus,
22};
23use util::{ResultExt, maybe};
24
25use crate::api::events::SnowflakeRow;
26use crate::db::billing_subscription::{
27 StripeCancellationReason, StripeSubscriptionStatus, SubscriptionKind,
28};
29use crate::llm::{DEFAULT_MAX_MONTHLY_SPEND, FREE_TIER_MONTHLY_SPENDING_LIMIT};
30use crate::rpc::{ResultExt as _, Server};
31use crate::{AppState, Cents, Error, Result};
32use crate::{db::UserId, llm::db::LlmDatabase};
33use crate::{
34 db::{
35 BillingSubscriptionId, CreateBillingCustomerParams, CreateBillingSubscriptionParams,
36 CreateProcessedStripeEventParams, UpdateBillingCustomerParams,
37 UpdateBillingPreferencesParams, UpdateBillingSubscriptionParams, billing_customer,
38 },
39 stripe_billing::StripeBilling,
40};
41
42pub fn router() -> Router {
43 Router::new()
44 .route(
45 "/billing/preferences",
46 get(get_billing_preferences).put(update_billing_preferences),
47 )
48 .route(
49 "/billing/subscriptions",
50 get(list_billing_subscriptions).post(create_billing_subscription),
51 )
52 .route(
53 "/billing/subscriptions/manage",
54 post(manage_billing_subscription),
55 )
56 .route("/billing/monthly_spend", get(get_monthly_spend))
57 .route("/billing/usage", get(get_current_usage))
58}
59
60#[derive(Debug, Deserialize)]
61struct GetBillingPreferencesParams {
62 github_user_id: i32,
63}
64
65#[derive(Debug, Serialize)]
66struct BillingPreferencesResponse {
67 max_monthly_llm_usage_spending_in_cents: i32,
68}
69
70async fn get_billing_preferences(
71 Extension(app): Extension<Arc<AppState>>,
72 Query(params): Query<GetBillingPreferencesParams>,
73) -> Result<Json<BillingPreferencesResponse>> {
74 let user = app
75 .db
76 .get_user_by_github_user_id(params.github_user_id)
77 .await?
78 .ok_or_else(|| anyhow!("user not found"))?;
79
80 let preferences = app.db.get_billing_preferences(user.id).await?;
81
82 Ok(Json(BillingPreferencesResponse {
83 max_monthly_llm_usage_spending_in_cents: preferences
84 .map_or(DEFAULT_MAX_MONTHLY_SPEND.0 as i32, |preferences| {
85 preferences.max_monthly_llm_usage_spending_in_cents
86 }),
87 }))
88}
89
90#[derive(Debug, Deserialize)]
91struct UpdateBillingPreferencesBody {
92 github_user_id: i32,
93 max_monthly_llm_usage_spending_in_cents: i32,
94}
95
96async fn update_billing_preferences(
97 Extension(app): Extension<Arc<AppState>>,
98 Extension(rpc_server): Extension<Arc<crate::rpc::Server>>,
99 extract::Json(body): extract::Json<UpdateBillingPreferencesBody>,
100) -> Result<Json<BillingPreferencesResponse>> {
101 let user = app
102 .db
103 .get_user_by_github_user_id(body.github_user_id)
104 .await?
105 .ok_or_else(|| anyhow!("user not found"))?;
106
107 let max_monthly_llm_usage_spending_in_cents =
108 body.max_monthly_llm_usage_spending_in_cents.max(0);
109
110 let billing_preferences =
111 if let Some(_billing_preferences) = app.db.get_billing_preferences(user.id).await? {
112 app.db
113 .update_billing_preferences(
114 user.id,
115 &UpdateBillingPreferencesParams {
116 max_monthly_llm_usage_spending_in_cents: ActiveValue::set(
117 max_monthly_llm_usage_spending_in_cents,
118 ),
119 },
120 )
121 .await?
122 } else {
123 app.db
124 .create_billing_preferences(
125 user.id,
126 &crate::db::CreateBillingPreferencesParams {
127 max_monthly_llm_usage_spending_in_cents,
128 },
129 )
130 .await?
131 };
132
133 SnowflakeRow::new(
134 "Spend Limit Updated",
135 Some(user.metrics_id),
136 user.admin,
137 None,
138 json!({
139 "user_id": user.id,
140 "max_monthly_llm_usage_spending_in_cents": billing_preferences.max_monthly_llm_usage_spending_in_cents,
141 }),
142 )
143 .write(&app.kinesis_client, &app.config.kinesis_stream)
144 .await
145 .log_err();
146
147 rpc_server.refresh_llm_tokens_for_user(user.id).await;
148
149 Ok(Json(BillingPreferencesResponse {
150 max_monthly_llm_usage_spending_in_cents: billing_preferences
151 .max_monthly_llm_usage_spending_in_cents,
152 }))
153}
154
155#[derive(Debug, Deserialize)]
156struct ListBillingSubscriptionsParams {
157 github_user_id: i32,
158}
159
160#[derive(Debug, Serialize)]
161struct BillingSubscriptionJson {
162 id: BillingSubscriptionId,
163 name: String,
164 status: StripeSubscriptionStatus,
165 trial_end_at: Option<String>,
166 cancel_at: Option<String>,
167 /// Whether this subscription can be canceled.
168 is_cancelable: bool,
169}
170
171#[derive(Debug, Serialize)]
172struct ListBillingSubscriptionsResponse {
173 subscriptions: Vec<BillingSubscriptionJson>,
174}
175
176async fn list_billing_subscriptions(
177 Extension(app): Extension<Arc<AppState>>,
178 Query(params): Query<ListBillingSubscriptionsParams>,
179) -> Result<Json<ListBillingSubscriptionsResponse>> {
180 let user = app
181 .db
182 .get_user_by_github_user_id(params.github_user_id)
183 .await?
184 .ok_or_else(|| anyhow!("user not found"))?;
185
186 let subscriptions = app.db.get_billing_subscriptions(user.id).await?;
187
188 Ok(Json(ListBillingSubscriptionsResponse {
189 subscriptions: subscriptions
190 .into_iter()
191 .map(|subscription| BillingSubscriptionJson {
192 id: subscription.id,
193 name: match subscription.kind {
194 Some(SubscriptionKind::ZedPro) => "Zed Pro".to_string(),
195 Some(SubscriptionKind::ZedProTrial) => "Zed Pro (Trial)".to_string(),
196 Some(SubscriptionKind::ZedFree) => "Zed Free".to_string(),
197 None => "Zed LLM Usage".to_string(),
198 },
199 status: subscription.stripe_subscription_status,
200 trial_end_at: if subscription.kind == Some(SubscriptionKind::ZedProTrial) {
201 maybe!({
202 let end_at = subscription.stripe_current_period_end?;
203 let end_at = DateTime::from_timestamp(end_at, 0)?;
204
205 Some(end_at.to_rfc3339_opts(SecondsFormat::Millis, true))
206 })
207 } else {
208 None
209 },
210 cancel_at: subscription.stripe_cancel_at.map(|cancel_at| {
211 cancel_at
212 .and_utc()
213 .to_rfc3339_opts(SecondsFormat::Millis, true)
214 }),
215 is_cancelable: subscription.stripe_subscription_status.is_cancelable()
216 && subscription.stripe_cancel_at.is_none(),
217 })
218 .collect(),
219 }))
220}
221
222#[derive(Debug, Clone, Copy, Deserialize)]
223#[serde(rename_all = "snake_case")]
224enum ProductCode {
225 ZedPro,
226 ZedProTrial,
227}
228
229#[derive(Debug, Deserialize)]
230struct CreateBillingSubscriptionBody {
231 github_user_id: i32,
232 product: Option<ProductCode>,
233}
234
235#[derive(Debug, Serialize)]
236struct CreateBillingSubscriptionResponse {
237 checkout_session_url: String,
238}
239
240/// Initiates a Stripe Checkout session for creating a billing subscription.
241async fn create_billing_subscription(
242 Extension(app): Extension<Arc<AppState>>,
243 extract::Json(body): extract::Json<CreateBillingSubscriptionBody>,
244) -> Result<Json<CreateBillingSubscriptionResponse>> {
245 let user = app
246 .db
247 .get_user_by_github_user_id(body.github_user_id)
248 .await?
249 .ok_or_else(|| anyhow!("user not found"))?;
250
251 let Some(stripe_client) = app.stripe_client.clone() else {
252 log::error!("failed to retrieve Stripe client");
253 Err(Error::http(
254 StatusCode::NOT_IMPLEMENTED,
255 "not supported".into(),
256 ))?
257 };
258 let Some(stripe_billing) = app.stripe_billing.clone() else {
259 log::error!("failed to retrieve Stripe billing object");
260 Err(Error::http(
261 StatusCode::NOT_IMPLEMENTED,
262 "not supported".into(),
263 ))?
264 };
265 let Some(llm_db) = app.llm_db.clone() else {
266 log::error!("failed to retrieve LLM database");
267 Err(Error::http(
268 StatusCode::NOT_IMPLEMENTED,
269 "not supported".into(),
270 ))?
271 };
272
273 if app.db.has_active_billing_subscription(user.id).await? {
274 return Err(Error::http(
275 StatusCode::CONFLICT,
276 "user already has an active subscription".into(),
277 ));
278 }
279
280 let existing_billing_customer = app.db.get_billing_customer_by_user_id(user.id).await?;
281 if let Some(existing_billing_customer) = &existing_billing_customer {
282 if existing_billing_customer.has_overdue_invoices {
283 return Err(Error::http(
284 StatusCode::PAYMENT_REQUIRED,
285 "user has overdue invoices".into(),
286 ));
287 }
288 }
289
290 let customer_id = if let Some(existing_customer) = existing_billing_customer {
291 CustomerId::from_str(&existing_customer.stripe_customer_id)
292 .context("failed to parse customer ID")?
293 } else {
294 let customer = Customer::create(
295 &stripe_client,
296 CreateCustomer {
297 email: user.email_address.as_deref(),
298 ..Default::default()
299 },
300 )
301 .await?;
302
303 customer.id
304 };
305
306 let success_url = format!(
307 "{}/account?checkout_complete=1",
308 app.config.zed_dot_dev_url()
309 );
310
311 let checkout_session_url = match body.product {
312 Some(ProductCode::ZedPro) => {
313 stripe_billing
314 .checkout_with_price(
315 app.config.zed_pro_price_id()?,
316 customer_id,
317 &user.github_login,
318 &success_url,
319 )
320 .await?
321 }
322 Some(ProductCode::ZedProTrial) => {
323 stripe_billing
324 .checkout_with_zed_pro_trial(
325 app.config.zed_pro_price_id()?,
326 customer_id,
327 &user.github_login,
328 &success_url,
329 )
330 .await?
331 }
332 None => {
333 let default_model = llm_db.model(
334 zed_llm_client::LanguageModelProvider::Anthropic,
335 "claude-3-7-sonnet",
336 )?;
337 let stripe_model = stripe_billing.register_model(default_model).await?;
338 stripe_billing
339 .checkout(customer_id, &user.github_login, &stripe_model, &success_url)
340 .await?
341 }
342 };
343
344 Ok(Json(CreateBillingSubscriptionResponse {
345 checkout_session_url,
346 }))
347}
348
349#[derive(Debug, PartialEq, Deserialize)]
350#[serde(rename_all = "snake_case")]
351enum ManageSubscriptionIntent {
352 /// The user intends to manage their subscription.
353 ///
354 /// This will open the Stripe billing portal without putting the user in a specific flow.
355 ManageSubscription,
356 /// The user intends to upgrade to Zed Pro.
357 UpgradeToPro,
358 /// The user intends to cancel their subscription.
359 Cancel,
360 /// The user intends to stop the cancellation of their subscription.
361 StopCancellation,
362}
363
364#[derive(Debug, Deserialize)]
365struct ManageBillingSubscriptionBody {
366 github_user_id: i32,
367 intent: ManageSubscriptionIntent,
368 /// The ID of the subscription to manage.
369 subscription_id: BillingSubscriptionId,
370}
371
372#[derive(Debug, Serialize)]
373struct ManageBillingSubscriptionResponse {
374 billing_portal_session_url: Option<String>,
375}
376
377/// Initiates a Stripe customer portal session for managing a billing subscription.
378async fn manage_billing_subscription(
379 Extension(app): Extension<Arc<AppState>>,
380 extract::Json(body): extract::Json<ManageBillingSubscriptionBody>,
381) -> Result<Json<ManageBillingSubscriptionResponse>> {
382 let user = app
383 .db
384 .get_user_by_github_user_id(body.github_user_id)
385 .await?
386 .ok_or_else(|| anyhow!("user not found"))?;
387
388 let Some(stripe_client) = app.stripe_client.clone() else {
389 log::error!("failed to retrieve Stripe client");
390 Err(Error::http(
391 StatusCode::NOT_IMPLEMENTED,
392 "not supported".into(),
393 ))?
394 };
395
396 let customer = app
397 .db
398 .get_billing_customer_by_user_id(user.id)
399 .await?
400 .ok_or_else(|| anyhow!("billing customer not found"))?;
401 let customer_id = CustomerId::from_str(&customer.stripe_customer_id)
402 .context("failed to parse customer ID")?;
403
404 let subscription = app
405 .db
406 .get_billing_subscription_by_id(body.subscription_id)
407 .await?
408 .ok_or_else(|| anyhow!("subscription not found"))?;
409 let subscription_id = SubscriptionId::from_str(&subscription.stripe_subscription_id)
410 .context("failed to parse subscription ID")?;
411
412 if body.intent == ManageSubscriptionIntent::StopCancellation {
413 let updated_stripe_subscription = Subscription::update(
414 &stripe_client,
415 &subscription_id,
416 stripe::UpdateSubscription {
417 cancel_at_period_end: Some(false),
418 ..Default::default()
419 },
420 )
421 .await?;
422
423 app.db
424 .update_billing_subscription(
425 subscription.id,
426 &UpdateBillingSubscriptionParams {
427 stripe_cancel_at: ActiveValue::set(
428 updated_stripe_subscription
429 .cancel_at
430 .and_then(|cancel_at| DateTime::from_timestamp(cancel_at, 0))
431 .map(|time| time.naive_utc()),
432 ),
433 ..Default::default()
434 },
435 )
436 .await?;
437
438 return Ok(Json(ManageBillingSubscriptionResponse {
439 billing_portal_session_url: None,
440 }));
441 }
442
443 let flow = match body.intent {
444 ManageSubscriptionIntent::ManageSubscription => None,
445 ManageSubscriptionIntent::UpgradeToPro => {
446 let zed_pro_price_id = app.config.zed_pro_price_id()?;
447 let zed_free_price_id = app.config.zed_free_price_id()?;
448
449 let stripe_subscription =
450 Subscription::retrieve(&stripe_client, &subscription_id, &[]).await?;
451
452 let is_on_zed_pro_trial = stripe_subscription.status == SubscriptionStatus::Trialing
453 && stripe_subscription.items.data.iter().any(|item| {
454 item.price
455 .as_ref()
456 .map_or(false, |price| price.id == zed_pro_price_id)
457 });
458 if is_on_zed_pro_trial {
459 // If the user is already on a Zed Pro trial and wants to upgrade to Pro, we just need to end their trial early.
460 Subscription::update(
461 &stripe_client,
462 &stripe_subscription.id,
463 stripe::UpdateSubscription {
464 trial_end: Some(stripe::Scheduled::now()),
465 ..Default::default()
466 },
467 )
468 .await?;
469
470 return Ok(Json(ManageBillingSubscriptionResponse {
471 billing_portal_session_url: None,
472 }));
473 }
474
475 let subscription_item_to_update = stripe_subscription
476 .items
477 .data
478 .iter()
479 .find_map(|item| {
480 let price = item.price.as_ref()?;
481
482 if price.id == zed_free_price_id {
483 Some(item.id.clone())
484 } else {
485 None
486 }
487 })
488 .ok_or_else(|| anyhow!("No subscription item to update"))?;
489
490 Some(CreateBillingPortalSessionFlowData {
491 type_: CreateBillingPortalSessionFlowDataType::SubscriptionUpdateConfirm,
492 subscription_update_confirm: Some(
493 CreateBillingPortalSessionFlowDataSubscriptionUpdateConfirm {
494 subscription: subscription.stripe_subscription_id,
495 items: vec![
496 CreateBillingPortalSessionFlowDataSubscriptionUpdateConfirmItems {
497 id: subscription_item_to_update.to_string(),
498 price: Some(zed_pro_price_id.to_string()),
499 quantity: Some(1),
500 },
501 ],
502 discounts: None,
503 },
504 ),
505 ..Default::default()
506 })
507 }
508 ManageSubscriptionIntent::Cancel => Some(CreateBillingPortalSessionFlowData {
509 type_: CreateBillingPortalSessionFlowDataType::SubscriptionCancel,
510 after_completion: Some(CreateBillingPortalSessionFlowDataAfterCompletion {
511 type_: stripe::CreateBillingPortalSessionFlowDataAfterCompletionType::Redirect,
512 redirect: Some(CreateBillingPortalSessionFlowDataAfterCompletionRedirect {
513 return_url: format!("{}/account", app.config.zed_dot_dev_url()),
514 }),
515 ..Default::default()
516 }),
517 subscription_cancel: Some(
518 stripe::CreateBillingPortalSessionFlowDataSubscriptionCancel {
519 subscription: subscription.stripe_subscription_id,
520 retention: None,
521 },
522 ),
523 ..Default::default()
524 }),
525 ManageSubscriptionIntent::StopCancellation => unreachable!(),
526 };
527
528 let mut params = CreateBillingPortalSession::new(customer_id);
529 params.flow_data = flow;
530 let return_url = format!("{}/account", app.config.zed_dot_dev_url());
531 params.return_url = Some(&return_url);
532
533 let session = BillingPortalSession::create(&stripe_client, params).await?;
534
535 Ok(Json(ManageBillingSubscriptionResponse {
536 billing_portal_session_url: Some(session.url),
537 }))
538}
539
540/// The amount of time we wait in between each poll of Stripe events.
541///
542/// This value should strike a balance between:
543/// 1. Being short enough that we update quickly when something in Stripe changes
544/// 2. Being long enough that we don't eat into our rate limits.
545///
546/// As a point of reference, the Sequin folks say they have this at **500ms**:
547///
548/// > We poll the Stripe /events endpoint every 500ms per account
549/// >
550/// > — https://blog.sequinstream.com/events-not-webhooks/
551const POLL_EVENTS_INTERVAL: Duration = Duration::from_secs(5);
552
553/// The maximum number of events to return per page.
554///
555/// We set this to 100 (the max) so we have to make fewer requests to Stripe.
556///
557/// > Limit can range between 1 and 100, and the default is 10.
558const EVENTS_LIMIT_PER_PAGE: u64 = 100;
559
560/// The number of pages consisting entirely of already-processed events that we
561/// will see before we stop retrieving events.
562///
563/// This is used to prevent over-fetching the Stripe events API for events we've
564/// already seen and processed.
565const NUMBER_OF_ALREADY_PROCESSED_PAGES_BEFORE_WE_STOP: usize = 4;
566
567/// Polls the Stripe events API periodically to reconcile the records in our
568/// database with the data in Stripe.
569pub fn poll_stripe_events_periodically(app: Arc<AppState>, rpc_server: Arc<Server>) {
570 let Some(stripe_client) = app.stripe_client.clone() else {
571 log::warn!("failed to retrieve Stripe client");
572 return;
573 };
574
575 let executor = app.executor.clone();
576 executor.spawn_detached({
577 let executor = executor.clone();
578 async move {
579 loop {
580 poll_stripe_events(&app, &rpc_server, &stripe_client)
581 .await
582 .log_err();
583
584 executor.sleep(POLL_EVENTS_INTERVAL).await;
585 }
586 }
587 });
588}
589
590async fn poll_stripe_events(
591 app: &Arc<AppState>,
592 rpc_server: &Arc<Server>,
593 stripe_client: &stripe::Client,
594) -> anyhow::Result<()> {
595 fn event_type_to_string(event_type: EventType) -> String {
596 // Calling `to_string` on `stripe::EventType` members gives us a quoted string,
597 // so we need to unquote it.
598 event_type.to_string().trim_matches('"').to_string()
599 }
600
601 let event_types = [
602 EventType::CustomerCreated,
603 EventType::CustomerUpdated,
604 EventType::CustomerSubscriptionCreated,
605 EventType::CustomerSubscriptionUpdated,
606 EventType::CustomerSubscriptionPaused,
607 EventType::CustomerSubscriptionResumed,
608 EventType::CustomerSubscriptionDeleted,
609 ]
610 .into_iter()
611 .map(event_type_to_string)
612 .collect::<Vec<_>>();
613
614 let mut pages_of_already_processed_events = 0;
615 let mut unprocessed_events = Vec::new();
616
617 log::info!(
618 "Stripe events: starting retrieval for {}",
619 event_types.join(", ")
620 );
621 let mut params = ListEvents::new();
622 params.types = Some(event_types.clone());
623 params.limit = Some(EVENTS_LIMIT_PER_PAGE);
624
625 let mut event_pages = stripe::Event::list(&stripe_client, ¶ms)
626 .await?
627 .paginate(params);
628
629 loop {
630 let processed_event_ids = {
631 let event_ids = event_pages
632 .page
633 .data
634 .iter()
635 .map(|event| event.id.as_str())
636 .collect::<Vec<_>>();
637 app.db
638 .get_processed_stripe_events_by_event_ids(&event_ids)
639 .await?
640 .into_iter()
641 .map(|event| event.stripe_event_id)
642 .collect::<Vec<_>>()
643 };
644
645 let mut processed_events_in_page = 0;
646 let events_in_page = event_pages.page.data.len();
647 for event in &event_pages.page.data {
648 if processed_event_ids.contains(&event.id.to_string()) {
649 processed_events_in_page += 1;
650 log::debug!("Stripe events: already processed '{}', skipping", event.id);
651 } else {
652 unprocessed_events.push(event.clone());
653 }
654 }
655
656 if processed_events_in_page == events_in_page {
657 pages_of_already_processed_events += 1;
658 }
659
660 if event_pages.page.has_more {
661 if pages_of_already_processed_events >= NUMBER_OF_ALREADY_PROCESSED_PAGES_BEFORE_WE_STOP
662 {
663 log::info!(
664 "Stripe events: stopping, saw {pages_of_already_processed_events} pages of already-processed events"
665 );
666 break;
667 } else {
668 log::info!("Stripe events: retrieving next page");
669 event_pages = event_pages.next(&stripe_client).await?;
670 }
671 } else {
672 break;
673 }
674 }
675
676 log::info!("Stripe events: unprocessed {}", unprocessed_events.len());
677
678 // Sort all of the unprocessed events in ascending order, so we can handle them in the order they occurred.
679 unprocessed_events.sort_by(|a, b| a.created.cmp(&b.created).then_with(|| a.id.cmp(&b.id)));
680
681 for event in unprocessed_events {
682 let event_id = event.id.clone();
683 let processed_event_params = CreateProcessedStripeEventParams {
684 stripe_event_id: event.id.to_string(),
685 stripe_event_type: event_type_to_string(event.type_),
686 stripe_event_created_timestamp: event.created,
687 };
688
689 // If the event has happened too far in the past, we don't want to
690 // process it and risk overwriting other more-recent updates.
691 //
692 // 1 day was chosen arbitrarily. This could be made longer or shorter.
693 let one_day = Duration::from_secs(24 * 60 * 60);
694 let a_day_ago = Utc::now() - one_day;
695 if a_day_ago.timestamp() > event.created {
696 log::info!(
697 "Stripe events: event '{}' is more than {one_day:?} old, marking as processed",
698 event_id
699 );
700 app.db
701 .create_processed_stripe_event(&processed_event_params)
702 .await?;
703
704 return Ok(());
705 }
706
707 let process_result = match event.type_ {
708 EventType::CustomerCreated | EventType::CustomerUpdated => {
709 handle_customer_event(app, stripe_client, event).await
710 }
711 EventType::CustomerSubscriptionCreated
712 | EventType::CustomerSubscriptionUpdated
713 | EventType::CustomerSubscriptionPaused
714 | EventType::CustomerSubscriptionResumed
715 | EventType::CustomerSubscriptionDeleted => {
716 handle_customer_subscription_event(app, rpc_server, stripe_client, event).await
717 }
718 _ => Ok(()),
719 };
720
721 if let Some(()) = process_result
722 .with_context(|| format!("failed to process event {event_id} successfully"))
723 .log_err()
724 {
725 app.db
726 .create_processed_stripe_event(&processed_event_params)
727 .await?;
728 }
729 }
730
731 Ok(())
732}
733
734async fn handle_customer_event(
735 app: &Arc<AppState>,
736 _stripe_client: &stripe::Client,
737 event: stripe::Event,
738) -> anyhow::Result<()> {
739 let EventObject::Customer(customer) = event.data.object else {
740 bail!("unexpected event payload for {}", event.id);
741 };
742
743 log::info!("handling Stripe {} event: {}", event.type_, event.id);
744
745 let Some(email) = customer.email else {
746 log::info!("Stripe customer has no email: skipping");
747 return Ok(());
748 };
749
750 let Some(user) = app.db.get_user_by_email(&email).await? else {
751 log::info!("no user found for email: skipping");
752 return Ok(());
753 };
754
755 if let Some(existing_customer) = app
756 .db
757 .get_billing_customer_by_stripe_customer_id(&customer.id)
758 .await?
759 {
760 app.db
761 .update_billing_customer(
762 existing_customer.id,
763 &UpdateBillingCustomerParams {
764 // For now we just leave the information as-is, as it is not
765 // likely to change.
766 ..Default::default()
767 },
768 )
769 .await?;
770 } else {
771 app.db
772 .create_billing_customer(&CreateBillingCustomerParams {
773 user_id: user.id,
774 stripe_customer_id: customer.id.to_string(),
775 })
776 .await?;
777 }
778
779 Ok(())
780}
781
782async fn handle_customer_subscription_event(
783 app: &Arc<AppState>,
784 rpc_server: &Arc<Server>,
785 stripe_client: &stripe::Client,
786 event: stripe::Event,
787) -> anyhow::Result<()> {
788 let EventObject::Subscription(subscription) = event.data.object else {
789 bail!("unexpected event payload for {}", event.id);
790 };
791
792 log::info!("handling Stripe {} event: {}", event.type_, event.id);
793
794 let subscription_kind = maybe!({
795 let zed_pro_price_id = app.config.zed_pro_price_id().ok()?;
796 let zed_free_price_id = app.config.zed_free_price_id().ok()?;
797
798 subscription.items.data.iter().find_map(|item| {
799 let price = item.price.as_ref()?;
800
801 if price.id == zed_pro_price_id {
802 Some(if subscription.status == SubscriptionStatus::Trialing {
803 SubscriptionKind::ZedProTrial
804 } else {
805 SubscriptionKind::ZedPro
806 })
807 } else if price.id == zed_free_price_id {
808 Some(SubscriptionKind::ZedFree)
809 } else {
810 None
811 }
812 })
813 });
814
815 let billing_customer =
816 find_or_create_billing_customer(app, stripe_client, subscription.customer)
817 .await?
818 .ok_or_else(|| anyhow!("billing customer not found"))?;
819
820 let was_canceled_due_to_payment_failure = subscription.status == SubscriptionStatus::Canceled
821 && subscription
822 .cancellation_details
823 .as_ref()
824 .and_then(|details| details.reason)
825 .map_or(false, |reason| {
826 reason == CancellationDetailsReason::PaymentFailed
827 });
828
829 if was_canceled_due_to_payment_failure {
830 app.db
831 .update_billing_customer(
832 billing_customer.id,
833 &UpdateBillingCustomerParams {
834 has_overdue_invoices: ActiveValue::set(true),
835 ..Default::default()
836 },
837 )
838 .await?;
839 }
840
841 if let Some(existing_subscription) = app
842 .db
843 .get_billing_subscription_by_stripe_subscription_id(&subscription.id)
844 .await?
845 {
846 let llm_db = app
847 .llm_db
848 .clone()
849 .ok_or_else(|| anyhow!("LLM DB not initialized"))?;
850
851 let new_period_start_at =
852 chrono::DateTime::from_timestamp(subscription.current_period_start, 0)
853 .ok_or_else(|| anyhow!("No subscription period start"))?;
854 let new_period_end_at =
855 chrono::DateTime::from_timestamp(subscription.current_period_end, 0)
856 .ok_or_else(|| anyhow!("No subscription period end"))?;
857
858 llm_db
859 .transfer_existing_subscription_usage(
860 billing_customer.user_id,
861 &existing_subscription,
862 subscription_kind,
863 new_period_start_at,
864 new_period_end_at,
865 )
866 .await?;
867
868 app.db
869 .update_billing_subscription(
870 existing_subscription.id,
871 &UpdateBillingSubscriptionParams {
872 billing_customer_id: ActiveValue::set(billing_customer.id),
873 kind: ActiveValue::set(subscription_kind),
874 stripe_subscription_id: ActiveValue::set(subscription.id.to_string()),
875 stripe_subscription_status: ActiveValue::set(subscription.status.into()),
876 stripe_cancel_at: ActiveValue::set(
877 subscription
878 .cancel_at
879 .and_then(|cancel_at| DateTime::from_timestamp(cancel_at, 0))
880 .map(|time| time.naive_utc()),
881 ),
882 stripe_cancellation_reason: ActiveValue::set(
883 subscription
884 .cancellation_details
885 .and_then(|details| details.reason)
886 .map(|reason| reason.into()),
887 ),
888 stripe_current_period_start: ActiveValue::set(Some(
889 subscription.current_period_start,
890 )),
891 stripe_current_period_end: ActiveValue::set(Some(
892 subscription.current_period_end,
893 )),
894 },
895 )
896 .await?;
897 } else {
898 // If the user already has an active billing subscription, ignore the
899 // event and return an `Ok` to signal that it was processed
900 // successfully.
901 //
902 // There is the possibility that this could cause us to not create a
903 // subscription in the following scenario:
904 //
905 // 1. User has an active subscription A
906 // 2. User cancels subscription A
907 // 3. User creates a new subscription B
908 // 4. We process the new subscription B before the cancellation of subscription A
909 // 5. User ends up with no subscriptions
910 //
911 // In theory this situation shouldn't arise as we try to process the events in the order they occur.
912 if app
913 .db
914 .has_active_billing_subscription(billing_customer.user_id)
915 .await?
916 {
917 log::info!(
918 "user {user_id} already has an active subscription, skipping creation of subscription {subscription_id}",
919 user_id = billing_customer.user_id,
920 subscription_id = subscription.id
921 );
922 return Ok(());
923 }
924
925 app.db
926 .create_billing_subscription(&CreateBillingSubscriptionParams {
927 billing_customer_id: billing_customer.id,
928 kind: subscription_kind,
929 stripe_subscription_id: subscription.id.to_string(),
930 stripe_subscription_status: subscription.status.into(),
931 stripe_cancellation_reason: subscription
932 .cancellation_details
933 .and_then(|details| details.reason)
934 .map(|reason| reason.into()),
935 stripe_current_period_start: Some(subscription.current_period_start),
936 stripe_current_period_end: Some(subscription.current_period_end),
937 })
938 .await?;
939 }
940
941 // When the user's subscription changes, we want to refresh their LLM tokens
942 // to either grant/revoke access.
943 rpc_server
944 .refresh_llm_tokens_for_user(billing_customer.user_id)
945 .await;
946
947 Ok(())
948}
949
950#[derive(Debug, Deserialize)]
951struct GetMonthlySpendParams {
952 github_user_id: i32,
953}
954
955#[derive(Debug, Serialize)]
956struct GetMonthlySpendResponse {
957 monthly_free_tier_spend_in_cents: u32,
958 monthly_free_tier_allowance_in_cents: u32,
959 monthly_spend_in_cents: u32,
960}
961
962async fn get_monthly_spend(
963 Extension(app): Extension<Arc<AppState>>,
964 Query(params): Query<GetMonthlySpendParams>,
965) -> Result<Json<GetMonthlySpendResponse>> {
966 let user = app
967 .db
968 .get_user_by_github_user_id(params.github_user_id)
969 .await?
970 .ok_or_else(|| anyhow!("user not found"))?;
971
972 let Some(llm_db) = app.llm_db.clone() else {
973 return Err(Error::http(
974 StatusCode::NOT_IMPLEMENTED,
975 "LLM database not available".into(),
976 ));
977 };
978
979 let free_tier = user
980 .custom_llm_monthly_allowance_in_cents
981 .map(|allowance| Cents(allowance as u32))
982 .unwrap_or(FREE_TIER_MONTHLY_SPENDING_LIMIT);
983
984 let spending_for_month = llm_db
985 .get_user_spending_for_month(user.id, Utc::now())
986 .await?;
987
988 let free_tier_spend = Cents::min(spending_for_month, free_tier);
989 let monthly_spend = spending_for_month.saturating_sub(free_tier);
990
991 Ok(Json(GetMonthlySpendResponse {
992 monthly_free_tier_spend_in_cents: free_tier_spend.0,
993 monthly_free_tier_allowance_in_cents: free_tier.0,
994 monthly_spend_in_cents: monthly_spend.0,
995 }))
996}
997
998#[derive(Debug, Deserialize)]
999struct GetCurrentUsageParams {
1000 github_user_id: i32,
1001}
1002
1003#[derive(Debug, Serialize)]
1004struct UsageCounts {
1005 pub used: i32,
1006 pub limit: Option<i32>,
1007 pub remaining: Option<i32>,
1008}
1009
1010#[derive(Debug, Serialize)]
1011struct GetCurrentUsageResponse {
1012 pub model_requests: UsageCounts,
1013 pub edit_predictions: UsageCounts,
1014}
1015
1016async fn get_current_usage(
1017 Extension(app): Extension<Arc<AppState>>,
1018 Query(params): Query<GetCurrentUsageParams>,
1019) -> Result<Json<GetCurrentUsageResponse>> {
1020 let user = app
1021 .db
1022 .get_user_by_github_user_id(params.github_user_id)
1023 .await?
1024 .ok_or_else(|| anyhow!("user not found"))?;
1025
1026 let Some(llm_db) = app.llm_db.clone() else {
1027 return Err(Error::http(
1028 StatusCode::NOT_IMPLEMENTED,
1029 "LLM database not available".into(),
1030 ));
1031 };
1032
1033 let empty_usage = GetCurrentUsageResponse {
1034 model_requests: UsageCounts {
1035 used: 0,
1036 limit: Some(0),
1037 remaining: Some(0),
1038 },
1039 edit_predictions: UsageCounts {
1040 used: 0,
1041 limit: Some(0),
1042 remaining: Some(0),
1043 },
1044 };
1045
1046 let Some(subscription) = app.db.get_active_billing_subscription(user.id).await? else {
1047 return Ok(Json(empty_usage));
1048 };
1049
1050 let subscription_period = maybe!({
1051 let period_start_at = subscription.current_period_start_at()?;
1052 let period_end_at = subscription.current_period_end_at()?;
1053
1054 Some((period_start_at, period_end_at))
1055 });
1056
1057 let Some((period_start_at, period_end_at)) = subscription_period else {
1058 return Ok(Json(empty_usage));
1059 };
1060
1061 let usage = llm_db
1062 .get_subscription_usage_for_period(user.id, period_start_at, period_end_at)
1063 .await?;
1064 let Some(usage) = usage else {
1065 return Ok(Json(empty_usage));
1066 };
1067
1068 let plan = match usage.plan {
1069 SubscriptionKind::ZedPro => zed_llm_client::Plan::ZedPro,
1070 SubscriptionKind::ZedProTrial => zed_llm_client::Plan::ZedProTrial,
1071 SubscriptionKind::ZedFree => zed_llm_client::Plan::Free,
1072 };
1073
1074 let model_requests_limit = match plan.model_requests_limit() {
1075 zed_llm_client::UsageLimit::Limited(limit) => Some(limit),
1076 zed_llm_client::UsageLimit::Unlimited => None,
1077 };
1078 let edit_prediction_limit = match plan.edit_predictions_limit() {
1079 zed_llm_client::UsageLimit::Limited(limit) => Some(limit),
1080 zed_llm_client::UsageLimit::Unlimited => None,
1081 };
1082
1083 Ok(Json(GetCurrentUsageResponse {
1084 model_requests: UsageCounts {
1085 used: usage.model_requests,
1086 limit: model_requests_limit,
1087 remaining: model_requests_limit.map(|limit| (limit - usage.model_requests).max(0)),
1088 },
1089 edit_predictions: UsageCounts {
1090 used: usage.edit_predictions,
1091 limit: edit_prediction_limit,
1092 remaining: edit_prediction_limit.map(|limit| (limit - usage.edit_predictions).max(0)),
1093 },
1094 }))
1095}
1096
1097impl From<SubscriptionStatus> for StripeSubscriptionStatus {
1098 fn from(value: SubscriptionStatus) -> Self {
1099 match value {
1100 SubscriptionStatus::Incomplete => Self::Incomplete,
1101 SubscriptionStatus::IncompleteExpired => Self::IncompleteExpired,
1102 SubscriptionStatus::Trialing => Self::Trialing,
1103 SubscriptionStatus::Active => Self::Active,
1104 SubscriptionStatus::PastDue => Self::PastDue,
1105 SubscriptionStatus::Canceled => Self::Canceled,
1106 SubscriptionStatus::Unpaid => Self::Unpaid,
1107 SubscriptionStatus::Paused => Self::Paused,
1108 }
1109 }
1110}
1111
1112impl From<CancellationDetailsReason> for StripeCancellationReason {
1113 fn from(value: CancellationDetailsReason) -> Self {
1114 match value {
1115 CancellationDetailsReason::CancellationRequested => Self::CancellationRequested,
1116 CancellationDetailsReason::PaymentDisputed => Self::PaymentDisputed,
1117 CancellationDetailsReason::PaymentFailed => Self::PaymentFailed,
1118 }
1119 }
1120}
1121
1122/// Finds or creates a billing customer using the provided customer.
1123async fn find_or_create_billing_customer(
1124 app: &Arc<AppState>,
1125 stripe_client: &stripe::Client,
1126 customer_or_id: Expandable<Customer>,
1127) -> anyhow::Result<Option<billing_customer::Model>> {
1128 let customer_id = match &customer_or_id {
1129 Expandable::Id(id) => id,
1130 Expandable::Object(customer) => customer.id.as_ref(),
1131 };
1132
1133 // If we already have a billing customer record associated with the Stripe customer,
1134 // there's nothing more we need to do.
1135 if let Some(billing_customer) = app
1136 .db
1137 .get_billing_customer_by_stripe_customer_id(customer_id)
1138 .await?
1139 {
1140 return Ok(Some(billing_customer));
1141 }
1142
1143 // If all we have is a customer ID, resolve it to a full customer record by
1144 // hitting the Stripe API.
1145 let customer = match customer_or_id {
1146 Expandable::Id(id) => Customer::retrieve(stripe_client, &id, &[]).await?,
1147 Expandable::Object(customer) => *customer,
1148 };
1149
1150 let Some(email) = customer.email else {
1151 return Ok(None);
1152 };
1153
1154 let Some(user) = app.db.get_user_by_email(&email).await? else {
1155 return Ok(None);
1156 };
1157
1158 let billing_customer = app
1159 .db
1160 .create_billing_customer(&CreateBillingCustomerParams {
1161 user_id: user.id,
1162 stripe_customer_id: customer.id.to_string(),
1163 })
1164 .await?;
1165
1166 Ok(Some(billing_customer))
1167}
1168
1169const SYNC_LLM_USAGE_WITH_STRIPE_INTERVAL: Duration = Duration::from_secs(60);
1170
1171pub fn sync_llm_usage_with_stripe_periodically(app: Arc<AppState>) {
1172 let Some(stripe_billing) = app.stripe_billing.clone() else {
1173 log::warn!("failed to retrieve Stripe billing object");
1174 return;
1175 };
1176 let Some(llm_db) = app.llm_db.clone() else {
1177 log::warn!("failed to retrieve LLM database");
1178 return;
1179 };
1180
1181 let executor = app.executor.clone();
1182 executor.spawn_detached({
1183 let executor = executor.clone();
1184 async move {
1185 loop {
1186 sync_with_stripe(&app, &llm_db, &stripe_billing)
1187 .await
1188 .context("failed to sync LLM usage to Stripe")
1189 .trace_err();
1190 executor.sleep(SYNC_LLM_USAGE_WITH_STRIPE_INTERVAL).await;
1191 }
1192 }
1193 });
1194}
1195
1196async fn sync_with_stripe(
1197 app: &Arc<AppState>,
1198 llm_db: &Arc<LlmDatabase>,
1199 stripe_billing: &Arc<StripeBilling>,
1200) -> anyhow::Result<()> {
1201 let events = llm_db.get_billing_events().await?;
1202 let user_ids = events
1203 .iter()
1204 .map(|(event, _)| event.user_id)
1205 .collect::<HashSet<UserId>>();
1206 let stripe_subscriptions = app.db.get_active_billing_subscriptions(user_ids).await?;
1207
1208 for (event, model) in events {
1209 let Some((stripe_db_customer, stripe_db_subscription)) =
1210 stripe_subscriptions.get(&event.user_id)
1211 else {
1212 tracing::warn!(
1213 user_id = event.user_id.0,
1214 "Registered billing event for user who is not a Stripe customer. Billing events should only be created for users who are Stripe customers, so this is a mistake on our side."
1215 );
1216 continue;
1217 };
1218 let stripe_subscription_id: stripe::SubscriptionId = stripe_db_subscription
1219 .stripe_subscription_id
1220 .parse()
1221 .context("failed to parse stripe subscription id from db")?;
1222 let stripe_customer_id: stripe::CustomerId = stripe_db_customer
1223 .stripe_customer_id
1224 .parse()
1225 .context("failed to parse stripe customer id from db")?;
1226
1227 let stripe_model = stripe_billing.register_model(&model).await?;
1228 stripe_billing
1229 .subscribe_to_model(&stripe_subscription_id, &stripe_model)
1230 .await?;
1231 stripe_billing
1232 .bill_model_usage(&stripe_customer_id, &stripe_model, &event)
1233 .await?;
1234 llm_db.consume_billing_event(event.id).await?;
1235 }
1236
1237 Ok(())
1238}