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