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