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