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