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