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