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, PaymentMethod, Subscription, SubscriptionId,
22 SubscriptionStatus,
23};
24use util::{ResultExt, maybe};
25
26use crate::api::events::SnowflakeRow;
27use crate::db::billing_subscription::{
28 StripeCancellationReason, StripeSubscriptionStatus, SubscriptionKind,
29};
30use crate::llm::db::subscription_usage_meter::CompletionMode;
31use crate::llm::{
32 AGENT_EXTENDED_TRIAL_FEATURE_FLAG, DEFAULT_MAX_MONTHLY_SPEND, FREE_TIER_MONTHLY_SPENDING_LIMIT,
33};
34use crate::rpc::{ResultExt as _, Server};
35use crate::{AppState, Cents, Error, Result};
36use crate::{db::UserId, llm::db::LlmDatabase};
37use crate::{
38 db::{
39 BillingSubscriptionId, CreateBillingCustomerParams, CreateBillingSubscriptionParams,
40 CreateProcessedStripeEventParams, UpdateBillingCustomerParams,
41 UpdateBillingPreferencesParams, UpdateBillingSubscriptionParams, billing_customer,
42 },
43 stripe_billing::StripeBilling,
44};
45
46pub fn router() -> Router {
47 Router::new()
48 .route(
49 "/billing/preferences",
50 get(get_billing_preferences).put(update_billing_preferences),
51 )
52 .route(
53 "/billing/subscriptions",
54 get(list_billing_subscriptions).post(create_billing_subscription),
55 )
56 .route(
57 "/billing/subscriptions/manage",
58 post(manage_billing_subscription),
59 )
60 .route(
61 "/billing/subscriptions/migrate",
62 post(migrate_to_new_billing),
63 )
64 .route(
65 "/billing/subscriptions/sync",
66 post(sync_billing_subscription),
67 )
68 .route("/billing/monthly_spend", get(get_monthly_spend))
69 .route("/billing/usage", get(get_current_usage))
70}
71
72#[derive(Debug, Deserialize)]
73struct GetBillingPreferencesParams {
74 github_user_id: i32,
75}
76
77#[derive(Debug, Serialize)]
78struct BillingPreferencesResponse {
79 trial_started_at: Option<String>,
80 max_monthly_llm_usage_spending_in_cents: i32,
81 model_request_overages_enabled: bool,
82 model_request_overages_spend_limit_in_cents: i32,
83}
84
85async fn get_billing_preferences(
86 Extension(app): Extension<Arc<AppState>>,
87 Query(params): Query<GetBillingPreferencesParams>,
88) -> Result<Json<BillingPreferencesResponse>> {
89 let user = app
90 .db
91 .get_user_by_github_user_id(params.github_user_id)
92 .await?
93 .ok_or_else(|| anyhow!("user not found"))?;
94
95 let billing_customer = app.db.get_billing_customer_by_user_id(user.id).await?;
96 let preferences = app.db.get_billing_preferences(user.id).await?;
97
98 Ok(Json(BillingPreferencesResponse {
99 trial_started_at: billing_customer
100 .and_then(|billing_customer| billing_customer.trial_started_at)
101 .map(|trial_started_at| {
102 trial_started_at
103 .and_utc()
104 .to_rfc3339_opts(SecondsFormat::Millis, true)
105 }),
106 max_monthly_llm_usage_spending_in_cents: preferences
107 .as_ref()
108 .map_or(DEFAULT_MAX_MONTHLY_SPEND.0 as i32, |preferences| {
109 preferences.max_monthly_llm_usage_spending_in_cents
110 }),
111 model_request_overages_enabled: preferences.as_ref().map_or(false, |preferences| {
112 preferences.model_request_overages_enabled
113 }),
114 model_request_overages_spend_limit_in_cents: preferences
115 .as_ref()
116 .map_or(0, |preferences| {
117 preferences.model_request_overages_spend_limit_in_cents
118 }),
119 }))
120}
121
122#[derive(Debug, Deserialize)]
123struct UpdateBillingPreferencesBody {
124 github_user_id: i32,
125 #[serde(default)]
126 max_monthly_llm_usage_spending_in_cents: i32,
127 #[serde(default)]
128 model_request_overages_enabled: bool,
129 #[serde(default)]
130 model_request_overages_spend_limit_in_cents: i32,
131}
132
133async fn update_billing_preferences(
134 Extension(app): Extension<Arc<AppState>>,
135 Extension(rpc_server): Extension<Arc<crate::rpc::Server>>,
136 extract::Json(body): extract::Json<UpdateBillingPreferencesBody>,
137) -> Result<Json<BillingPreferencesResponse>> {
138 let user = app
139 .db
140 .get_user_by_github_user_id(body.github_user_id)
141 .await?
142 .ok_or_else(|| anyhow!("user not found"))?;
143
144 let billing_customer = app.db.get_billing_customer_by_user_id(user.id).await?;
145
146 let max_monthly_llm_usage_spending_in_cents =
147 body.max_monthly_llm_usage_spending_in_cents.max(0);
148 let model_request_overages_spend_limit_in_cents =
149 body.model_request_overages_spend_limit_in_cents.max(0);
150
151 let billing_preferences =
152 if let Some(_billing_preferences) = app.db.get_billing_preferences(user.id).await? {
153 app.db
154 .update_billing_preferences(
155 user.id,
156 &UpdateBillingPreferencesParams {
157 max_monthly_llm_usage_spending_in_cents: ActiveValue::set(
158 max_monthly_llm_usage_spending_in_cents,
159 ),
160 model_request_overages_enabled: ActiveValue::set(
161 body.model_request_overages_enabled,
162 ),
163 model_request_overages_spend_limit_in_cents: ActiveValue::set(
164 model_request_overages_spend_limit_in_cents,
165 ),
166 },
167 )
168 .await?
169 } else {
170 app.db
171 .create_billing_preferences(
172 user.id,
173 &crate::db::CreateBillingPreferencesParams {
174 max_monthly_llm_usage_spending_in_cents,
175 model_request_overages_enabled: body.model_request_overages_enabled,
176 model_request_overages_spend_limit_in_cents,
177 },
178 )
179 .await?
180 };
181
182 SnowflakeRow::new(
183 "Billing Preferences Updated",
184 Some(user.metrics_id),
185 user.admin,
186 None,
187 json!({
188 "user_id": user.id,
189 "model_request_overages_enabled": billing_preferences.model_request_overages_enabled,
190 "model_request_overages_spend_limit_in_cents": billing_preferences.model_request_overages_spend_limit_in_cents,
191 "max_monthly_llm_usage_spending_in_cents": billing_preferences.max_monthly_llm_usage_spending_in_cents,
192 }),
193 )
194 .write(&app.kinesis_client, &app.config.kinesis_stream)
195 .await
196 .log_err();
197
198 rpc_server.refresh_llm_tokens_for_user(user.id).await;
199
200 Ok(Json(BillingPreferencesResponse {
201 trial_started_at: billing_customer
202 .and_then(|billing_customer| billing_customer.trial_started_at)
203 .map(|trial_started_at| {
204 trial_started_at
205 .and_utc()
206 .to_rfc3339_opts(SecondsFormat::Millis, true)
207 }),
208 max_monthly_llm_usage_spending_in_cents: billing_preferences
209 .max_monthly_llm_usage_spending_in_cents,
210 model_request_overages_enabled: billing_preferences.model_request_overages_enabled,
211 model_request_overages_spend_limit_in_cents: billing_preferences
212 .model_request_overages_spend_limit_in_cents,
213 }))
214}
215
216#[derive(Debug, Deserialize)]
217struct ListBillingSubscriptionsParams {
218 github_user_id: i32,
219}
220
221#[derive(Debug, Serialize)]
222struct BillingSubscriptionJson {
223 id: BillingSubscriptionId,
224 name: String,
225 status: StripeSubscriptionStatus,
226 trial_end_at: Option<String>,
227 cancel_at: Option<String>,
228 /// Whether this subscription can be canceled.
229 is_cancelable: bool,
230}
231
232#[derive(Debug, Serialize)]
233struct ListBillingSubscriptionsResponse {
234 subscriptions: Vec<BillingSubscriptionJson>,
235}
236
237async fn list_billing_subscriptions(
238 Extension(app): Extension<Arc<AppState>>,
239 Query(params): Query<ListBillingSubscriptionsParams>,
240) -> Result<Json<ListBillingSubscriptionsResponse>> {
241 let user = app
242 .db
243 .get_user_by_github_user_id(params.github_user_id)
244 .await?
245 .ok_or_else(|| anyhow!("user not found"))?;
246
247 let subscriptions = app.db.get_billing_subscriptions(user.id).await?;
248
249 Ok(Json(ListBillingSubscriptionsResponse {
250 subscriptions: subscriptions
251 .into_iter()
252 .map(|subscription| BillingSubscriptionJson {
253 id: subscription.id,
254 name: match subscription.kind {
255 Some(SubscriptionKind::ZedPro) => "Zed Pro".to_string(),
256 Some(SubscriptionKind::ZedProTrial) => "Zed Pro (Trial)".to_string(),
257 Some(SubscriptionKind::ZedFree) => "Zed Free".to_string(),
258 None => "Zed LLM Usage".to_string(),
259 },
260 status: subscription.stripe_subscription_status,
261 trial_end_at: if subscription.kind == Some(SubscriptionKind::ZedProTrial) {
262 maybe!({
263 let end_at = subscription.stripe_current_period_end?;
264 let end_at = DateTime::from_timestamp(end_at, 0)?;
265
266 Some(end_at.to_rfc3339_opts(SecondsFormat::Millis, true))
267 })
268 } else {
269 None
270 },
271 cancel_at: subscription.stripe_cancel_at.map(|cancel_at| {
272 cancel_at
273 .and_utc()
274 .to_rfc3339_opts(SecondsFormat::Millis, true)
275 }),
276 is_cancelable: subscription.stripe_subscription_status.is_cancelable()
277 && subscription.stripe_cancel_at.is_none(),
278 })
279 .collect(),
280 }))
281}
282
283#[derive(Debug, Clone, Copy, Deserialize)]
284#[serde(rename_all = "snake_case")]
285enum ProductCode {
286 ZedPro,
287 ZedProTrial,
288 ZedFree,
289}
290
291#[derive(Debug, Deserialize)]
292struct CreateBillingSubscriptionBody {
293 github_user_id: i32,
294 product: Option<ProductCode>,
295}
296
297#[derive(Debug, Serialize)]
298struct CreateBillingSubscriptionResponse {
299 checkout_session_url: String,
300}
301
302/// Initiates a Stripe Checkout session for creating a billing subscription.
303async fn create_billing_subscription(
304 Extension(app): Extension<Arc<AppState>>,
305 extract::Json(body): extract::Json<CreateBillingSubscriptionBody>,
306) -> Result<Json<CreateBillingSubscriptionResponse>> {
307 let user = app
308 .db
309 .get_user_by_github_user_id(body.github_user_id)
310 .await?
311 .ok_or_else(|| anyhow!("user not found"))?;
312
313 let Some(stripe_client) = app.stripe_client.clone() else {
314 log::error!("failed to retrieve Stripe client");
315 Err(Error::http(
316 StatusCode::NOT_IMPLEMENTED,
317 "not supported".into(),
318 ))?
319 };
320 let Some(stripe_billing) = app.stripe_billing.clone() else {
321 log::error!("failed to retrieve Stripe billing object");
322 Err(Error::http(
323 StatusCode::NOT_IMPLEMENTED,
324 "not supported".into(),
325 ))?
326 };
327
328 if app.db.has_active_billing_subscription(user.id).await? {
329 return Err(Error::http(
330 StatusCode::CONFLICT,
331 "user already has an active subscription".into(),
332 ));
333 }
334
335 let existing_billing_customer = app.db.get_billing_customer_by_user_id(user.id).await?;
336 if let Some(existing_billing_customer) = &existing_billing_customer {
337 if existing_billing_customer.has_overdue_invoices {
338 return Err(Error::http(
339 StatusCode::PAYMENT_REQUIRED,
340 "user has overdue invoices".into(),
341 ));
342 }
343 }
344
345 let customer_id = if let Some(existing_customer) = &existing_billing_customer {
346 CustomerId::from_str(&existing_customer.stripe_customer_id)
347 .context("failed to parse customer ID")?
348 } else {
349 let existing_customer = if let Some(email) = user.email_address.as_deref() {
350 let customers = Customer::list(
351 &stripe_client,
352 &stripe::ListCustomers {
353 email: Some(email),
354 ..Default::default()
355 },
356 )
357 .await?;
358
359 customers.data.first().cloned()
360 } else {
361 None
362 };
363
364 if let Some(existing_customer) = existing_customer {
365 existing_customer.id
366 } else {
367 let customer = Customer::create(
368 &stripe_client,
369 CreateCustomer {
370 email: user.email_address.as_deref(),
371 ..Default::default()
372 },
373 )
374 .await?;
375
376 customer.id
377 }
378 };
379
380 let success_url = format!(
381 "{}/account?checkout_complete=1",
382 app.config.zed_dot_dev_url()
383 );
384
385 let checkout_session_url = match body.product {
386 Some(ProductCode::ZedPro) => {
387 stripe_billing
388 .checkout_with_zed_pro(customer_id, &user.github_login, &success_url)
389 .await?
390 }
391 Some(ProductCode::ZedProTrial) => {
392 if let Some(existing_billing_customer) = &existing_billing_customer {
393 if existing_billing_customer.trial_started_at.is_some() {
394 return Err(Error::http(
395 StatusCode::FORBIDDEN,
396 "user already used free trial".into(),
397 ));
398 }
399 }
400
401 let feature_flags = app.db.get_user_flags(user.id).await?;
402
403 stripe_billing
404 .checkout_with_zed_pro_trial(
405 customer_id,
406 &user.github_login,
407 feature_flags,
408 &success_url,
409 )
410 .await?
411 }
412 Some(ProductCode::ZedFree) => {
413 stripe_billing
414 .checkout_with_zed_free(customer_id, &user.github_login, &success_url)
415 .await?
416 }
417 None => {
418 return Err(Error::http(
419 StatusCode::BAD_REQUEST,
420 "No product selected".into(),
421 ));
422 }
423 };
424
425 Ok(Json(CreateBillingSubscriptionResponse {
426 checkout_session_url,
427 }))
428}
429
430#[derive(Debug, PartialEq, Deserialize)]
431#[serde(rename_all = "snake_case")]
432enum ManageSubscriptionIntent {
433 /// The user intends to manage their subscription.
434 ///
435 /// This will open the Stripe billing portal without putting the user in a specific flow.
436 ManageSubscription,
437 /// The user intends to update their payment method.
438 UpdatePaymentMethod,
439 /// The user intends to upgrade to Zed Pro.
440 UpgradeToPro,
441 /// The user intends to cancel their subscription.
442 Cancel,
443 /// The user intends to stop the cancellation of their subscription.
444 StopCancellation,
445}
446
447#[derive(Debug, Deserialize)]
448struct ManageBillingSubscriptionBody {
449 github_user_id: i32,
450 intent: ManageSubscriptionIntent,
451 /// The ID of the subscription to manage.
452 subscription_id: BillingSubscriptionId,
453 redirect_to: Option<String>,
454}
455
456#[derive(Debug, Serialize)]
457struct ManageBillingSubscriptionResponse {
458 billing_portal_session_url: Option<String>,
459}
460
461/// Initiates a Stripe customer portal session for managing a billing subscription.
462async fn manage_billing_subscription(
463 Extension(app): Extension<Arc<AppState>>,
464 extract::Json(body): extract::Json<ManageBillingSubscriptionBody>,
465) -> Result<Json<ManageBillingSubscriptionResponse>> {
466 let user = app
467 .db
468 .get_user_by_github_user_id(body.github_user_id)
469 .await?
470 .ok_or_else(|| anyhow!("user not found"))?;
471
472 let Some(stripe_client) = app.stripe_client.clone() else {
473 log::error!("failed to retrieve Stripe client");
474 Err(Error::http(
475 StatusCode::NOT_IMPLEMENTED,
476 "not supported".into(),
477 ))?
478 };
479
480 let Some(stripe_billing) = app.stripe_billing.clone() else {
481 log::error!("failed to retrieve Stripe billing object");
482 Err(Error::http(
483 StatusCode::NOT_IMPLEMENTED,
484 "not supported".into(),
485 ))?
486 };
487
488 let customer = app
489 .db
490 .get_billing_customer_by_user_id(user.id)
491 .await?
492 .ok_or_else(|| anyhow!("billing customer not found"))?;
493 let customer_id = CustomerId::from_str(&customer.stripe_customer_id)
494 .context("failed to parse customer ID")?;
495
496 let subscription = app
497 .db
498 .get_billing_subscription_by_id(body.subscription_id)
499 .await?
500 .ok_or_else(|| anyhow!("subscription not found"))?;
501 let subscription_id = SubscriptionId::from_str(&subscription.stripe_subscription_id)
502 .context("failed to parse subscription ID")?;
503
504 if body.intent == ManageSubscriptionIntent::StopCancellation {
505 let updated_stripe_subscription = Subscription::update(
506 &stripe_client,
507 &subscription_id,
508 stripe::UpdateSubscription {
509 cancel_at_period_end: Some(false),
510 ..Default::default()
511 },
512 )
513 .await?;
514
515 app.db
516 .update_billing_subscription(
517 subscription.id,
518 &UpdateBillingSubscriptionParams {
519 stripe_cancel_at: ActiveValue::set(
520 updated_stripe_subscription
521 .cancel_at
522 .and_then(|cancel_at| DateTime::from_timestamp(cancel_at, 0))
523 .map(|time| time.naive_utc()),
524 ),
525 ..Default::default()
526 },
527 )
528 .await?;
529
530 return Ok(Json(ManageBillingSubscriptionResponse {
531 billing_portal_session_url: None,
532 }));
533 }
534
535 let flow = match body.intent {
536 ManageSubscriptionIntent::ManageSubscription => None,
537 ManageSubscriptionIntent::UpgradeToPro => {
538 let zed_pro_price_id = stripe_billing.zed_pro_price_id().await?;
539 let zed_free_price_id = stripe_billing.zed_free_price_id().await?;
540
541 let stripe_subscription =
542 Subscription::retrieve(&stripe_client, &subscription_id, &[]).await?;
543
544 let is_on_zed_pro_trial = stripe_subscription.status == SubscriptionStatus::Trialing
545 && stripe_subscription.items.data.iter().any(|item| {
546 item.price
547 .as_ref()
548 .map_or(false, |price| price.id == zed_pro_price_id)
549 });
550 if is_on_zed_pro_trial {
551 let payment_methods = PaymentMethod::list(
552 &stripe_client,
553 &stripe::ListPaymentMethods {
554 customer: Some(stripe_subscription.customer.id()),
555 ..Default::default()
556 },
557 )
558 .await?;
559
560 let has_payment_method = !payment_methods.data.is_empty();
561 if !has_payment_method {
562 return Err(Error::http(
563 StatusCode::BAD_REQUEST,
564 "missing payment method".into(),
565 ));
566 }
567
568 // If the user is already on a Zed Pro trial and wants to upgrade to Pro, we just need to end their trial early.
569 Subscription::update(
570 &stripe_client,
571 &stripe_subscription.id,
572 stripe::UpdateSubscription {
573 trial_end: Some(stripe::Scheduled::now()),
574 ..Default::default()
575 },
576 )
577 .await?;
578
579 return Ok(Json(ManageBillingSubscriptionResponse {
580 billing_portal_session_url: None,
581 }));
582 }
583
584 let subscription_item_to_update = stripe_subscription
585 .items
586 .data
587 .iter()
588 .find_map(|item| {
589 let price = item.price.as_ref()?;
590
591 if price.id == zed_free_price_id {
592 Some(item.id.clone())
593 } else {
594 None
595 }
596 })
597 .ok_or_else(|| anyhow!("No subscription item to update"))?;
598
599 Some(CreateBillingPortalSessionFlowData {
600 type_: CreateBillingPortalSessionFlowDataType::SubscriptionUpdateConfirm,
601 subscription_update_confirm: Some(
602 CreateBillingPortalSessionFlowDataSubscriptionUpdateConfirm {
603 subscription: subscription.stripe_subscription_id,
604 items: vec![
605 CreateBillingPortalSessionFlowDataSubscriptionUpdateConfirmItems {
606 id: subscription_item_to_update.to_string(),
607 price: Some(zed_pro_price_id.to_string()),
608 quantity: Some(1),
609 },
610 ],
611 discounts: None,
612 },
613 ),
614 ..Default::default()
615 })
616 }
617 ManageSubscriptionIntent::UpdatePaymentMethod => Some(CreateBillingPortalSessionFlowData {
618 type_: CreateBillingPortalSessionFlowDataType::PaymentMethodUpdate,
619 after_completion: Some(CreateBillingPortalSessionFlowDataAfterCompletion {
620 type_: stripe::CreateBillingPortalSessionFlowDataAfterCompletionType::Redirect,
621 redirect: Some(CreateBillingPortalSessionFlowDataAfterCompletionRedirect {
622 return_url: format!(
623 "{}{path}",
624 app.config.zed_dot_dev_url(),
625 path = body.redirect_to.unwrap_or_else(|| "/account".to_string())
626 ),
627 }),
628 ..Default::default()
629 }),
630 ..Default::default()
631 }),
632 ManageSubscriptionIntent::Cancel => Some(CreateBillingPortalSessionFlowData {
633 type_: CreateBillingPortalSessionFlowDataType::SubscriptionCancel,
634 after_completion: Some(CreateBillingPortalSessionFlowDataAfterCompletion {
635 type_: stripe::CreateBillingPortalSessionFlowDataAfterCompletionType::Redirect,
636 redirect: Some(CreateBillingPortalSessionFlowDataAfterCompletionRedirect {
637 return_url: format!("{}/account", app.config.zed_dot_dev_url()),
638 }),
639 ..Default::default()
640 }),
641 subscription_cancel: Some(
642 stripe::CreateBillingPortalSessionFlowDataSubscriptionCancel {
643 subscription: subscription.stripe_subscription_id,
644 retention: None,
645 },
646 ),
647 ..Default::default()
648 }),
649 ManageSubscriptionIntent::StopCancellation => unreachable!(),
650 };
651
652 let mut params = CreateBillingPortalSession::new(customer_id);
653 params.flow_data = flow;
654 let return_url = format!("{}/account", app.config.zed_dot_dev_url());
655 params.return_url = Some(&return_url);
656
657 let session = BillingPortalSession::create(&stripe_client, params).await?;
658
659 Ok(Json(ManageBillingSubscriptionResponse {
660 billing_portal_session_url: Some(session.url),
661 }))
662}
663
664#[derive(Debug, Deserialize)]
665struct MigrateToNewBillingBody {
666 github_user_id: i32,
667}
668
669#[derive(Debug, Serialize)]
670struct MigrateToNewBillingResponse {
671 /// The ID of the subscription that was canceled.
672 canceled_subscription_id: Option<String>,
673}
674
675async fn migrate_to_new_billing(
676 Extension(app): Extension<Arc<AppState>>,
677 extract::Json(body): extract::Json<MigrateToNewBillingBody>,
678) -> Result<Json<MigrateToNewBillingResponse>> {
679 let Some(stripe_client) = app.stripe_client.clone() else {
680 log::error!("failed to retrieve Stripe client");
681 Err(Error::http(
682 StatusCode::NOT_IMPLEMENTED,
683 "not supported".into(),
684 ))?
685 };
686
687 let user = app
688 .db
689 .get_user_by_github_user_id(body.github_user_id)
690 .await?
691 .ok_or_else(|| anyhow!("user not found"))?;
692
693 let old_billing_subscriptions_by_user = app
694 .db
695 .get_active_billing_subscriptions(HashSet::from_iter([user.id]))
696 .await?;
697
698 let canceled_subscription_id = if let Some((_billing_customer, billing_subscription)) =
699 old_billing_subscriptions_by_user.get(&user.id)
700 {
701 let stripe_subscription_id = billing_subscription
702 .stripe_subscription_id
703 .parse::<stripe::SubscriptionId>()
704 .context("failed to parse Stripe subscription ID from database")?;
705
706 Subscription::cancel(
707 &stripe_client,
708 &stripe_subscription_id,
709 stripe::CancelSubscription {
710 invoice_now: Some(true),
711 ..Default::default()
712 },
713 )
714 .await?;
715
716 Some(stripe_subscription_id)
717 } else {
718 None
719 };
720
721 let all_feature_flags = app.db.list_feature_flags().await?;
722 let user_feature_flags = app.db.get_user_flags(user.id).await?;
723
724 for feature_flag in ["new-billing", "assistant2"] {
725 let already_in_feature_flag = user_feature_flags.iter().any(|flag| flag == feature_flag);
726 if already_in_feature_flag {
727 continue;
728 }
729
730 let feature_flag = all_feature_flags
731 .iter()
732 .find(|flag| flag.flag == feature_flag)
733 .context("failed to find feature flag: {feature_flag:?}")?;
734
735 app.db.add_user_flag(user.id, feature_flag.id).await?;
736 }
737
738 Ok(Json(MigrateToNewBillingResponse {
739 canceled_subscription_id: canceled_subscription_id
740 .map(|subscription_id| subscription_id.to_string()),
741 }))
742}
743
744#[derive(Debug, Deserialize)]
745struct SyncBillingSubscriptionBody {
746 github_user_id: i32,
747}
748
749#[derive(Debug, Serialize)]
750struct SyncBillingSubscriptionResponse {
751 stripe_customer_id: String,
752}
753
754async fn sync_billing_subscription(
755 Extension(app): Extension<Arc<AppState>>,
756 extract::Json(body): extract::Json<SyncBillingSubscriptionBody>,
757) -> Result<Json<SyncBillingSubscriptionResponse>> {
758 let Some(stripe_client) = app.stripe_client.clone() else {
759 log::error!("failed to retrieve Stripe client");
760 Err(Error::http(
761 StatusCode::NOT_IMPLEMENTED,
762 "not supported".into(),
763 ))?
764 };
765
766 let user = app
767 .db
768 .get_user_by_github_user_id(body.github_user_id)
769 .await?
770 .ok_or_else(|| anyhow!("user not found"))?;
771
772 let billing_customer = app
773 .db
774 .get_billing_customer_by_user_id(user.id)
775 .await?
776 .ok_or_else(|| anyhow!("billing customer not found"))?;
777 let stripe_customer_id = billing_customer
778 .stripe_customer_id
779 .parse::<stripe::CustomerId>()
780 .context("failed to parse Stripe customer ID from database")?;
781
782 let subscriptions = Subscription::list(
783 &stripe_client,
784 &stripe::ListSubscriptions {
785 customer: Some(stripe_customer_id),
786 // Sync all non-canceled subscriptions.
787 status: None,
788 ..Default::default()
789 },
790 )
791 .await?;
792
793 for subscription in subscriptions.data {
794 let subscription_id = subscription.id.clone();
795
796 sync_subscription(&app, &stripe_client, subscription)
797 .await
798 .with_context(|| {
799 format!(
800 "failed to sync subscription {subscription_id} for user {}",
801 user.id,
802 )
803 })?;
804 }
805
806 Ok(Json(SyncBillingSubscriptionResponse {
807 stripe_customer_id: billing_customer.stripe_customer_id.clone(),
808 }))
809}
810
811/// The amount of time we wait in between each poll of Stripe events.
812///
813/// This value should strike a balance between:
814/// 1. Being short enough that we update quickly when something in Stripe changes
815/// 2. Being long enough that we don't eat into our rate limits.
816///
817/// As a point of reference, the Sequin folks say they have this at **500ms**:
818///
819/// > We poll the Stripe /events endpoint every 500ms per account
820/// >
821/// > — https://blog.sequinstream.com/events-not-webhooks/
822const POLL_EVENTS_INTERVAL: Duration = Duration::from_secs(5);
823
824/// The maximum number of events to return per page.
825///
826/// We set this to 100 (the max) so we have to make fewer requests to Stripe.
827///
828/// > Limit can range between 1 and 100, and the default is 10.
829const EVENTS_LIMIT_PER_PAGE: u64 = 100;
830
831/// The number of pages consisting entirely of already-processed events that we
832/// will see before we stop retrieving events.
833///
834/// This is used to prevent over-fetching the Stripe events API for events we've
835/// already seen and processed.
836const NUMBER_OF_ALREADY_PROCESSED_PAGES_BEFORE_WE_STOP: usize = 4;
837
838/// Polls the Stripe events API periodically to reconcile the records in our
839/// database with the data in Stripe.
840pub fn poll_stripe_events_periodically(app: Arc<AppState>, rpc_server: Arc<Server>) {
841 let Some(stripe_client) = app.stripe_client.clone() else {
842 log::warn!("failed to retrieve Stripe client");
843 return;
844 };
845
846 let executor = app.executor.clone();
847 executor.spawn_detached({
848 let executor = executor.clone();
849 async move {
850 loop {
851 poll_stripe_events(&app, &rpc_server, &stripe_client)
852 .await
853 .log_err();
854
855 executor.sleep(POLL_EVENTS_INTERVAL).await;
856 }
857 }
858 });
859}
860
861async fn poll_stripe_events(
862 app: &Arc<AppState>,
863 rpc_server: &Arc<Server>,
864 stripe_client: &stripe::Client,
865) -> anyhow::Result<()> {
866 fn event_type_to_string(event_type: EventType) -> String {
867 // Calling `to_string` on `stripe::EventType` members gives us a quoted string,
868 // so we need to unquote it.
869 event_type.to_string().trim_matches('"').to_string()
870 }
871
872 let event_types = [
873 EventType::CustomerCreated,
874 EventType::CustomerUpdated,
875 EventType::CustomerSubscriptionCreated,
876 EventType::CustomerSubscriptionUpdated,
877 EventType::CustomerSubscriptionPaused,
878 EventType::CustomerSubscriptionResumed,
879 EventType::CustomerSubscriptionDeleted,
880 ]
881 .into_iter()
882 .map(event_type_to_string)
883 .collect::<Vec<_>>();
884
885 let mut pages_of_already_processed_events = 0;
886 let mut unprocessed_events = Vec::new();
887
888 log::info!(
889 "Stripe events: starting retrieval for {}",
890 event_types.join(", ")
891 );
892 let mut params = ListEvents::new();
893 params.types = Some(event_types.clone());
894 params.limit = Some(EVENTS_LIMIT_PER_PAGE);
895
896 let mut event_pages = stripe::Event::list(&stripe_client, ¶ms)
897 .await?
898 .paginate(params);
899
900 loop {
901 let processed_event_ids = {
902 let event_ids = event_pages
903 .page
904 .data
905 .iter()
906 .map(|event| event.id.as_str())
907 .collect::<Vec<_>>();
908 app.db
909 .get_processed_stripe_events_by_event_ids(&event_ids)
910 .await?
911 .into_iter()
912 .map(|event| event.stripe_event_id)
913 .collect::<Vec<_>>()
914 };
915
916 let mut processed_events_in_page = 0;
917 let events_in_page = event_pages.page.data.len();
918 for event in &event_pages.page.data {
919 if processed_event_ids.contains(&event.id.to_string()) {
920 processed_events_in_page += 1;
921 log::debug!("Stripe events: already processed '{}', skipping", event.id);
922 } else {
923 unprocessed_events.push(event.clone());
924 }
925 }
926
927 if processed_events_in_page == events_in_page {
928 pages_of_already_processed_events += 1;
929 }
930
931 if event_pages.page.has_more {
932 if pages_of_already_processed_events >= NUMBER_OF_ALREADY_PROCESSED_PAGES_BEFORE_WE_STOP
933 {
934 log::info!(
935 "Stripe events: stopping, saw {pages_of_already_processed_events} pages of already-processed events"
936 );
937 break;
938 } else {
939 log::info!("Stripe events: retrieving next page");
940 event_pages = event_pages.next(&stripe_client).await?;
941 }
942 } else {
943 break;
944 }
945 }
946
947 log::info!("Stripe events: unprocessed {}", unprocessed_events.len());
948
949 // Sort all of the unprocessed events in ascending order, so we can handle them in the order they occurred.
950 unprocessed_events.sort_by(|a, b| a.created.cmp(&b.created).then_with(|| a.id.cmp(&b.id)));
951
952 for event in unprocessed_events {
953 let event_id = event.id.clone();
954 let processed_event_params = CreateProcessedStripeEventParams {
955 stripe_event_id: event.id.to_string(),
956 stripe_event_type: event_type_to_string(event.type_),
957 stripe_event_created_timestamp: event.created,
958 };
959
960 // If the event has happened too far in the past, we don't want to
961 // process it and risk overwriting other more-recent updates.
962 //
963 // 1 day was chosen arbitrarily. This could be made longer or shorter.
964 let one_day = Duration::from_secs(24 * 60 * 60);
965 let a_day_ago = Utc::now() - one_day;
966 if a_day_ago.timestamp() > event.created {
967 log::info!(
968 "Stripe events: event '{}' is more than {one_day:?} old, marking as processed",
969 event_id
970 );
971 app.db
972 .create_processed_stripe_event(&processed_event_params)
973 .await?;
974
975 return Ok(());
976 }
977
978 let process_result = match event.type_ {
979 EventType::CustomerCreated | EventType::CustomerUpdated => {
980 handle_customer_event(app, stripe_client, event).await
981 }
982 EventType::CustomerSubscriptionCreated
983 | EventType::CustomerSubscriptionUpdated
984 | EventType::CustomerSubscriptionPaused
985 | EventType::CustomerSubscriptionResumed
986 | EventType::CustomerSubscriptionDeleted => {
987 handle_customer_subscription_event(app, rpc_server, stripe_client, event).await
988 }
989 _ => Ok(()),
990 };
991
992 if let Some(()) = process_result
993 .with_context(|| format!("failed to process event {event_id} successfully"))
994 .log_err()
995 {
996 app.db
997 .create_processed_stripe_event(&processed_event_params)
998 .await?;
999 }
1000 }
1001
1002 Ok(())
1003}
1004
1005async fn handle_customer_event(
1006 app: &Arc<AppState>,
1007 _stripe_client: &stripe::Client,
1008 event: stripe::Event,
1009) -> anyhow::Result<()> {
1010 let EventObject::Customer(customer) = event.data.object else {
1011 bail!("unexpected event payload for {}", event.id);
1012 };
1013
1014 log::info!("handling Stripe {} event: {}", event.type_, event.id);
1015
1016 let Some(email) = customer.email else {
1017 log::info!("Stripe customer has no email: skipping");
1018 return Ok(());
1019 };
1020
1021 let Some(user) = app.db.get_user_by_email(&email).await? else {
1022 log::info!("no user found for email: skipping");
1023 return Ok(());
1024 };
1025
1026 if let Some(existing_customer) = app
1027 .db
1028 .get_billing_customer_by_stripe_customer_id(&customer.id)
1029 .await?
1030 {
1031 app.db
1032 .update_billing_customer(
1033 existing_customer.id,
1034 &UpdateBillingCustomerParams {
1035 // For now we just leave the information as-is, as it is not
1036 // likely to change.
1037 ..Default::default()
1038 },
1039 )
1040 .await?;
1041 } else {
1042 app.db
1043 .create_billing_customer(&CreateBillingCustomerParams {
1044 user_id: user.id,
1045 stripe_customer_id: customer.id.to_string(),
1046 })
1047 .await?;
1048 }
1049
1050 Ok(())
1051}
1052
1053async fn sync_subscription(
1054 app: &Arc<AppState>,
1055 stripe_client: &stripe::Client,
1056 subscription: stripe::Subscription,
1057) -> anyhow::Result<billing_customer::Model> {
1058 let subscription_kind = if let Some(stripe_billing) = &app.stripe_billing {
1059 stripe_billing
1060 .determine_subscription_kind(&subscription)
1061 .await
1062 } else {
1063 None
1064 };
1065
1066 let billing_customer =
1067 find_or_create_billing_customer(app, stripe_client, subscription.customer)
1068 .await?
1069 .ok_or_else(|| anyhow!("billing customer not found"))?;
1070
1071 if let Some(SubscriptionKind::ZedProTrial) = subscription_kind {
1072 if subscription.status == SubscriptionStatus::Trialing {
1073 let current_period_start =
1074 DateTime::from_timestamp(subscription.current_period_start, 0)
1075 .ok_or_else(|| anyhow!("No trial subscription period start"))?;
1076
1077 app.db
1078 .update_billing_customer(
1079 billing_customer.id,
1080 &UpdateBillingCustomerParams {
1081 trial_started_at: ActiveValue::set(Some(current_period_start.naive_utc())),
1082 ..Default::default()
1083 },
1084 )
1085 .await?;
1086 }
1087 }
1088
1089 let was_canceled_due_to_payment_failure = subscription.status == SubscriptionStatus::Canceled
1090 && subscription
1091 .cancellation_details
1092 .as_ref()
1093 .and_then(|details| details.reason)
1094 .map_or(false, |reason| {
1095 reason == CancellationDetailsReason::PaymentFailed
1096 });
1097
1098 if was_canceled_due_to_payment_failure {
1099 app.db
1100 .update_billing_customer(
1101 billing_customer.id,
1102 &UpdateBillingCustomerParams {
1103 has_overdue_invoices: ActiveValue::set(true),
1104 ..Default::default()
1105 },
1106 )
1107 .await?;
1108 }
1109
1110 if let Some(existing_subscription) = app
1111 .db
1112 .get_billing_subscription_by_stripe_subscription_id(&subscription.id)
1113 .await?
1114 {
1115 app.db
1116 .update_billing_subscription(
1117 existing_subscription.id,
1118 &UpdateBillingSubscriptionParams {
1119 billing_customer_id: ActiveValue::set(billing_customer.id),
1120 kind: ActiveValue::set(subscription_kind),
1121 stripe_subscription_id: ActiveValue::set(subscription.id.to_string()),
1122 stripe_subscription_status: ActiveValue::set(subscription.status.into()),
1123 stripe_cancel_at: ActiveValue::set(
1124 subscription
1125 .cancel_at
1126 .and_then(|cancel_at| DateTime::from_timestamp(cancel_at, 0))
1127 .map(|time| time.naive_utc()),
1128 ),
1129 stripe_cancellation_reason: ActiveValue::set(
1130 subscription
1131 .cancellation_details
1132 .and_then(|details| details.reason)
1133 .map(|reason| reason.into()),
1134 ),
1135 stripe_current_period_start: ActiveValue::set(Some(
1136 subscription.current_period_start,
1137 )),
1138 stripe_current_period_end: ActiveValue::set(Some(
1139 subscription.current_period_end,
1140 )),
1141 },
1142 )
1143 .await?;
1144 } else {
1145 // If the user already has an active billing subscription, ignore the
1146 // event and return an `Ok` to signal that it was processed
1147 // successfully.
1148 //
1149 // There is the possibility that this could cause us to not create a
1150 // subscription in the following scenario:
1151 //
1152 // 1. User has an active subscription A
1153 // 2. User cancels subscription A
1154 // 3. User creates a new subscription B
1155 // 4. We process the new subscription B before the cancellation of subscription A
1156 // 5. User ends up with no subscriptions
1157 //
1158 // In theory this situation shouldn't arise as we try to process the events in the order they occur.
1159 if app
1160 .db
1161 .has_active_billing_subscription(billing_customer.user_id)
1162 .await?
1163 {
1164 log::info!(
1165 "user {user_id} already has an active subscription, skipping creation of subscription {subscription_id}",
1166 user_id = billing_customer.user_id,
1167 subscription_id = subscription.id
1168 );
1169 return Ok(billing_customer);
1170 }
1171
1172 app.db
1173 .create_billing_subscription(&CreateBillingSubscriptionParams {
1174 billing_customer_id: billing_customer.id,
1175 kind: subscription_kind,
1176 stripe_subscription_id: subscription.id.to_string(),
1177 stripe_subscription_status: subscription.status.into(),
1178 stripe_cancellation_reason: subscription
1179 .cancellation_details
1180 .and_then(|details| details.reason)
1181 .map(|reason| reason.into()),
1182 stripe_current_period_start: Some(subscription.current_period_start),
1183 stripe_current_period_end: Some(subscription.current_period_end),
1184 })
1185 .await?;
1186 }
1187
1188 Ok(billing_customer)
1189}
1190
1191async fn handle_customer_subscription_event(
1192 app: &Arc<AppState>,
1193 rpc_server: &Arc<Server>,
1194 stripe_client: &stripe::Client,
1195 event: stripe::Event,
1196) -> anyhow::Result<()> {
1197 let EventObject::Subscription(subscription) = event.data.object else {
1198 bail!("unexpected event payload for {}", event.id);
1199 };
1200
1201 log::info!("handling Stripe {} event: {}", event.type_, event.id);
1202
1203 let billing_customer = sync_subscription(app, stripe_client, subscription).await?;
1204
1205 // When the user's subscription changes, push down any changes to their plan.
1206 rpc_server
1207 .update_plan_for_user(billing_customer.user_id)
1208 .await
1209 .trace_err();
1210
1211 // When the user's subscription changes, we want to refresh their LLM tokens
1212 // to either grant/revoke access.
1213 rpc_server
1214 .refresh_llm_tokens_for_user(billing_customer.user_id)
1215 .await;
1216
1217 Ok(())
1218}
1219
1220#[derive(Debug, Deserialize)]
1221struct GetMonthlySpendParams {
1222 github_user_id: i32,
1223}
1224
1225#[derive(Debug, Serialize)]
1226struct GetMonthlySpendResponse {
1227 monthly_free_tier_spend_in_cents: u32,
1228 monthly_free_tier_allowance_in_cents: u32,
1229 monthly_spend_in_cents: u32,
1230}
1231
1232async fn get_monthly_spend(
1233 Extension(app): Extension<Arc<AppState>>,
1234 Query(params): Query<GetMonthlySpendParams>,
1235) -> Result<Json<GetMonthlySpendResponse>> {
1236 let user = app
1237 .db
1238 .get_user_by_github_user_id(params.github_user_id)
1239 .await?
1240 .ok_or_else(|| anyhow!("user not found"))?;
1241
1242 let Some(llm_db) = app.llm_db.clone() else {
1243 return Err(Error::http(
1244 StatusCode::NOT_IMPLEMENTED,
1245 "LLM database not available".into(),
1246 ));
1247 };
1248
1249 let free_tier = user
1250 .custom_llm_monthly_allowance_in_cents
1251 .map(|allowance| Cents(allowance as u32))
1252 .unwrap_or(FREE_TIER_MONTHLY_SPENDING_LIMIT);
1253
1254 let spending_for_month = llm_db
1255 .get_user_spending_for_month(user.id, Utc::now())
1256 .await?;
1257
1258 let free_tier_spend = Cents::min(spending_for_month, free_tier);
1259 let monthly_spend = spending_for_month.saturating_sub(free_tier);
1260
1261 Ok(Json(GetMonthlySpendResponse {
1262 monthly_free_tier_spend_in_cents: free_tier_spend.0,
1263 monthly_free_tier_allowance_in_cents: free_tier.0,
1264 monthly_spend_in_cents: monthly_spend.0,
1265 }))
1266}
1267
1268#[derive(Debug, Deserialize)]
1269struct GetCurrentUsageParams {
1270 github_user_id: i32,
1271}
1272
1273#[derive(Debug, Serialize)]
1274struct UsageCounts {
1275 pub used: i32,
1276 pub limit: Option<i32>,
1277 pub remaining: Option<i32>,
1278}
1279
1280#[derive(Debug, Serialize)]
1281struct ModelRequestUsage {
1282 pub model: String,
1283 pub mode: CompletionMode,
1284 pub requests: i32,
1285}
1286
1287#[derive(Debug, Serialize)]
1288struct CurrentUsage {
1289 pub model_requests: UsageCounts,
1290 pub model_request_usage: Vec<ModelRequestUsage>,
1291 pub edit_predictions: UsageCounts,
1292}
1293
1294#[derive(Debug, Default, Serialize)]
1295struct GetCurrentUsageResponse {
1296 pub plan: String,
1297 pub current_usage: Option<CurrentUsage>,
1298}
1299
1300async fn get_current_usage(
1301 Extension(app): Extension<Arc<AppState>>,
1302 Query(params): Query<GetCurrentUsageParams>,
1303) -> Result<Json<GetCurrentUsageResponse>> {
1304 let user = app
1305 .db
1306 .get_user_by_github_user_id(params.github_user_id)
1307 .await?
1308 .ok_or_else(|| anyhow!("user not found"))?;
1309
1310 let feature_flags = app.db.get_user_flags(user.id).await?;
1311 let has_extended_trial = feature_flags
1312 .iter()
1313 .any(|flag| flag == AGENT_EXTENDED_TRIAL_FEATURE_FLAG);
1314
1315 let Some(llm_db) = app.llm_db.clone() else {
1316 return Err(Error::http(
1317 StatusCode::NOT_IMPLEMENTED,
1318 "LLM database not available".into(),
1319 ));
1320 };
1321
1322 let Some(subscription) = app.db.get_active_billing_subscription(user.id).await? else {
1323 return Ok(Json(GetCurrentUsageResponse::default()));
1324 };
1325
1326 let subscription_period = maybe!({
1327 let period_start_at = subscription.current_period_start_at()?;
1328 let period_end_at = subscription.current_period_end_at()?;
1329
1330 Some((period_start_at, period_end_at))
1331 });
1332
1333 let Some((period_start_at, period_end_at)) = subscription_period else {
1334 return Ok(Json(GetCurrentUsageResponse::default()));
1335 };
1336
1337 let usage = llm_db
1338 .get_subscription_usage_for_period(user.id, period_start_at, period_end_at)
1339 .await?;
1340
1341 let plan = usage
1342 .as_ref()
1343 .map(|usage| usage.plan.into())
1344 .unwrap_or_else(|| {
1345 subscription
1346 .kind
1347 .map(Into::into)
1348 .unwrap_or(zed_llm_client::Plan::ZedFree)
1349 });
1350
1351 let model_requests_limit = match plan.model_requests_limit() {
1352 zed_llm_client::UsageLimit::Limited(limit) => {
1353 let limit = if plan == zed_llm_client::Plan::ZedProTrial && has_extended_trial {
1354 1_000
1355 } else {
1356 limit
1357 };
1358
1359 Some(limit)
1360 }
1361 zed_llm_client::UsageLimit::Unlimited => None,
1362 };
1363
1364 let edit_predictions_limit = match plan.edit_predictions_limit() {
1365 zed_llm_client::UsageLimit::Limited(limit) => Some(limit),
1366 zed_llm_client::UsageLimit::Unlimited => None,
1367 };
1368
1369 let Some(usage) = usage else {
1370 return Ok(Json(GetCurrentUsageResponse {
1371 plan: plan.as_str().to_string(),
1372 current_usage: Some(CurrentUsage {
1373 model_requests: UsageCounts {
1374 used: 0,
1375 limit: model_requests_limit,
1376 remaining: model_requests_limit,
1377 },
1378 model_request_usage: Vec::new(),
1379 edit_predictions: UsageCounts {
1380 used: 0,
1381 limit: edit_predictions_limit,
1382 remaining: edit_predictions_limit,
1383 },
1384 }),
1385 }));
1386 };
1387
1388 let subscription_usage_meters = llm_db
1389 .get_current_subscription_usage_meters_for_user(user.id, Utc::now())
1390 .await?;
1391
1392 let model_request_usage = subscription_usage_meters
1393 .into_iter()
1394 .filter_map(|(usage_meter, _usage)| {
1395 let model = llm_db.model_by_id(usage_meter.model_id).ok()?;
1396
1397 Some(ModelRequestUsage {
1398 model: model.name.clone(),
1399 mode: usage_meter.mode,
1400 requests: usage_meter.requests,
1401 })
1402 })
1403 .collect::<Vec<_>>();
1404
1405 Ok(Json(GetCurrentUsageResponse {
1406 plan: plan.as_str().to_string(),
1407 current_usage: Some(CurrentUsage {
1408 model_requests: UsageCounts {
1409 used: usage.model_requests,
1410 limit: model_requests_limit,
1411 remaining: model_requests_limit.map(|limit| (limit - usage.model_requests).max(0)),
1412 },
1413 model_request_usage,
1414 edit_predictions: UsageCounts {
1415 used: usage.edit_predictions,
1416 limit: edit_predictions_limit,
1417 remaining: edit_predictions_limit
1418 .map(|limit| (limit - usage.edit_predictions).max(0)),
1419 },
1420 }),
1421 }))
1422}
1423
1424impl From<SubscriptionStatus> for StripeSubscriptionStatus {
1425 fn from(value: SubscriptionStatus) -> Self {
1426 match value {
1427 SubscriptionStatus::Incomplete => Self::Incomplete,
1428 SubscriptionStatus::IncompleteExpired => Self::IncompleteExpired,
1429 SubscriptionStatus::Trialing => Self::Trialing,
1430 SubscriptionStatus::Active => Self::Active,
1431 SubscriptionStatus::PastDue => Self::PastDue,
1432 SubscriptionStatus::Canceled => Self::Canceled,
1433 SubscriptionStatus::Unpaid => Self::Unpaid,
1434 SubscriptionStatus::Paused => Self::Paused,
1435 }
1436 }
1437}
1438
1439impl From<CancellationDetailsReason> for StripeCancellationReason {
1440 fn from(value: CancellationDetailsReason) -> Self {
1441 match value {
1442 CancellationDetailsReason::CancellationRequested => Self::CancellationRequested,
1443 CancellationDetailsReason::PaymentDisputed => Self::PaymentDisputed,
1444 CancellationDetailsReason::PaymentFailed => Self::PaymentFailed,
1445 }
1446 }
1447}
1448
1449/// Finds or creates a billing customer using the provided customer.
1450async fn find_or_create_billing_customer(
1451 app: &Arc<AppState>,
1452 stripe_client: &stripe::Client,
1453 customer_or_id: Expandable<Customer>,
1454) -> anyhow::Result<Option<billing_customer::Model>> {
1455 let customer_id = match &customer_or_id {
1456 Expandable::Id(id) => id,
1457 Expandable::Object(customer) => customer.id.as_ref(),
1458 };
1459
1460 // If we already have a billing customer record associated with the Stripe customer,
1461 // there's nothing more we need to do.
1462 if let Some(billing_customer) = app
1463 .db
1464 .get_billing_customer_by_stripe_customer_id(customer_id)
1465 .await?
1466 {
1467 return Ok(Some(billing_customer));
1468 }
1469
1470 // If all we have is a customer ID, resolve it to a full customer record by
1471 // hitting the Stripe API.
1472 let customer = match customer_or_id {
1473 Expandable::Id(id) => Customer::retrieve(stripe_client, &id, &[]).await?,
1474 Expandable::Object(customer) => *customer,
1475 };
1476
1477 let Some(email) = customer.email else {
1478 return Ok(None);
1479 };
1480
1481 let Some(user) = app.db.get_user_by_email(&email).await? else {
1482 return Ok(None);
1483 };
1484
1485 let billing_customer = app
1486 .db
1487 .create_billing_customer(&CreateBillingCustomerParams {
1488 user_id: user.id,
1489 stripe_customer_id: customer.id.to_string(),
1490 })
1491 .await?;
1492
1493 Ok(Some(billing_customer))
1494}
1495
1496const SYNC_LLM_REQUEST_USAGE_WITH_STRIPE_INTERVAL: Duration = Duration::from_secs(60);
1497
1498pub fn sync_llm_request_usage_with_stripe_periodically(app: Arc<AppState>) {
1499 let Some(stripe_billing) = app.stripe_billing.clone() else {
1500 log::warn!("failed to retrieve Stripe billing object");
1501 return;
1502 };
1503 let Some(llm_db) = app.llm_db.clone() else {
1504 log::warn!("failed to retrieve LLM database");
1505 return;
1506 };
1507
1508 let executor = app.executor.clone();
1509 executor.spawn_detached({
1510 let executor = executor.clone();
1511 async move {
1512 loop {
1513 sync_model_request_usage_with_stripe(&app, &llm_db, &stripe_billing)
1514 .await
1515 .context("failed to sync LLM request usage to Stripe")
1516 .trace_err();
1517 executor
1518 .sleep(SYNC_LLM_REQUEST_USAGE_WITH_STRIPE_INTERVAL)
1519 .await;
1520 }
1521 }
1522 });
1523}
1524
1525async fn sync_model_request_usage_with_stripe(
1526 app: &Arc<AppState>,
1527 llm_db: &Arc<LlmDatabase>,
1528 stripe_billing: &Arc<StripeBilling>,
1529) -> anyhow::Result<()> {
1530 let staff_users = app.db.get_staff_users().await?;
1531 let staff_user_ids = staff_users
1532 .iter()
1533 .map(|user| user.id)
1534 .collect::<HashSet<UserId>>();
1535
1536 let usage_meters = llm_db
1537 .get_current_subscription_usage_meters(Utc::now())
1538 .await?;
1539 let usage_meters = usage_meters
1540 .into_iter()
1541 .filter(|(_, usage)| !staff_user_ids.contains(&usage.user_id))
1542 .collect::<Vec<_>>();
1543 let user_ids = usage_meters
1544 .iter()
1545 .map(|(_, usage)| usage.user_id)
1546 .collect::<HashSet<UserId>>();
1547 let billing_subscriptions = app
1548 .db
1549 .get_active_zed_pro_billing_subscriptions(user_ids)
1550 .await?;
1551
1552 let claude_3_5_sonnet = stripe_billing
1553 .find_price_by_lookup_key("claude-3-5-sonnet-requests")
1554 .await?;
1555 let claude_3_7_sonnet = stripe_billing
1556 .find_price_by_lookup_key("claude-3-7-sonnet-requests")
1557 .await?;
1558 let claude_3_7_sonnet_max = stripe_billing
1559 .find_price_by_lookup_key("claude-3-7-sonnet-requests-max")
1560 .await?;
1561
1562 for (usage_meter, usage) in usage_meters {
1563 maybe!(async {
1564 let Some((billing_customer, billing_subscription)) =
1565 billing_subscriptions.get(&usage.user_id)
1566 else {
1567 bail!(
1568 "Attempted to sync usage meter for user who is not a Stripe customer: {}",
1569 usage.user_id
1570 );
1571 };
1572
1573 let stripe_customer_id = billing_customer
1574 .stripe_customer_id
1575 .parse::<stripe::CustomerId>()
1576 .context("failed to parse Stripe customer ID from database")?;
1577 let stripe_subscription_id = billing_subscription
1578 .stripe_subscription_id
1579 .parse::<stripe::SubscriptionId>()
1580 .context("failed to parse Stripe subscription ID from database")?;
1581
1582 let model = llm_db.model_by_id(usage_meter.model_id)?;
1583
1584 let (price, meter_event_name) = match model.name.as_str() {
1585 "claude-3-5-sonnet" => (&claude_3_5_sonnet, "claude_3_5_sonnet/requests"),
1586 "claude-3-7-sonnet" => match usage_meter.mode {
1587 CompletionMode::Normal => (&claude_3_7_sonnet, "claude_3_7_sonnet/requests"),
1588 CompletionMode::Max => {
1589 (&claude_3_7_sonnet_max, "claude_3_7_sonnet/requests/max")
1590 }
1591 },
1592 model_name => {
1593 bail!("Attempted to sync usage meter for unsupported model: {model_name:?}")
1594 }
1595 };
1596
1597 stripe_billing
1598 .subscribe_to_price(&stripe_subscription_id, price)
1599 .await?;
1600 stripe_billing
1601 .bill_model_request_usage(
1602 &stripe_customer_id,
1603 meter_event_name,
1604 usage_meter.requests,
1605 )
1606 .await?;
1607
1608 Ok(())
1609 })
1610 .await
1611 .log_err();
1612 }
1613
1614 Ok(())
1615}