1use anyhow::{anyhow, bail, Context};
2use axum::{
3 extract::{self, Query},
4 routing::{get, post},
5 Extension, Json, Router,
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, CreateBillingPortalSession, CreateBillingPortalSessionFlowData,
16 CreateBillingPortalSessionFlowDataAfterCompletion,
17 CreateBillingPortalSessionFlowDataAfterCompletionRedirect,
18 CreateBillingPortalSessionFlowDataType, CreateCustomer, Customer, CustomerId, EventObject,
19 EventType, Expandable, ListEvents, Subscription, SubscriptionId, SubscriptionStatus,
20};
21use util::ResultExt;
22
23use crate::api::events::SnowflakeRow;
24use crate::llm::{DEFAULT_MAX_MONTHLY_SPEND, FREE_TIER_MONTHLY_SPENDING_LIMIT};
25use crate::rpc::{ResultExt as _, Server};
26use crate::{
27 db::{
28 billing_customer, BillingSubscriptionId, CreateBillingCustomerParams,
29 CreateBillingSubscriptionParams, CreateProcessedStripeEventParams,
30 UpdateBillingCustomerParams, UpdateBillingPreferencesParams,
31 UpdateBillingSubscriptionParams,
32 },
33 stripe_billing::StripeBilling,
34};
35use crate::{
36 db::{billing_subscription::StripeSubscriptionStatus, UserId},
37 llm::db::LlmDatabase,
38};
39use crate::{AppState, Cents, Error, Result};
40
41pub fn router() -> Router {
42 Router::new()
43 .route(
44 "/billing/preferences",
45 get(get_billing_preferences).put(update_billing_preferences),
46 )
47 .route(
48 "/billing/subscriptions",
49 get(list_billing_subscriptions).post(create_billing_subscription),
50 )
51 .route(
52 "/billing/subscriptions/manage",
53 post(manage_billing_subscription),
54 )
55 .route("/billing/monthly_spend", get(get_monthly_spend))
56}
57
58#[derive(Debug, Deserialize)]
59struct GetBillingPreferencesParams {
60 github_user_id: i32,
61}
62
63#[derive(Debug, Serialize)]
64struct BillingPreferencesResponse {
65 max_monthly_llm_usage_spending_in_cents: i32,
66}
67
68async fn get_billing_preferences(
69 Extension(app): Extension<Arc<AppState>>,
70 Query(params): Query<GetBillingPreferencesParams>,
71) -> Result<Json<BillingPreferencesResponse>> {
72 let user = app
73 .db
74 .get_user_by_github_user_id(params.github_user_id)
75 .await?
76 .ok_or_else(|| anyhow!("user not found"))?;
77
78 let preferences = app.db.get_billing_preferences(user.id).await?;
79
80 Ok(Json(BillingPreferencesResponse {
81 max_monthly_llm_usage_spending_in_cents: preferences
82 .map_or(DEFAULT_MAX_MONTHLY_SPEND.0 as i32, |preferences| {
83 preferences.max_monthly_llm_usage_spending_in_cents
84 }),
85 }))
86}
87
88#[derive(Debug, Deserialize)]
89struct UpdateBillingPreferencesBody {
90 github_user_id: i32,
91 max_monthly_llm_usage_spending_in_cents: i32,
92}
93
94async fn update_billing_preferences(
95 Extension(app): Extension<Arc<AppState>>,
96 Extension(rpc_server): Extension<Arc<crate::rpc::Server>>,
97 extract::Json(body): extract::Json<UpdateBillingPreferencesBody>,
98) -> Result<Json<BillingPreferencesResponse>> {
99 let user = app
100 .db
101 .get_user_by_github_user_id(body.github_user_id)
102 .await?
103 .ok_or_else(|| anyhow!("user not found"))?;
104
105 let max_monthly_llm_usage_spending_in_cents =
106 body.max_monthly_llm_usage_spending_in_cents.max(0);
107
108 let billing_preferences =
109 if let Some(_billing_preferences) = app.db.get_billing_preferences(user.id).await? {
110 app.db
111 .update_billing_preferences(
112 user.id,
113 &UpdateBillingPreferencesParams {
114 max_monthly_llm_usage_spending_in_cents: ActiveValue::set(
115 max_monthly_llm_usage_spending_in_cents,
116 ),
117 },
118 )
119 .await?
120 } else {
121 app.db
122 .create_billing_preferences(
123 user.id,
124 &crate::db::CreateBillingPreferencesParams {
125 max_monthly_llm_usage_spending_in_cents,
126 },
127 )
128 .await?
129 };
130
131 SnowflakeRow::new(
132 "Spend Limit Updated",
133 user.metrics_id,
134 user.admin,
135 None,
136 json!({
137 "user_id": user.id,
138 "max_monthly_llm_usage_spending_in_cents": billing_preferences.max_monthly_llm_usage_spending_in_cents,
139 }),
140 )
141 .write(&app.kinesis_client, &app.config.kinesis_stream)
142 .await
143 .log_err();
144
145 rpc_server.refresh_llm_tokens_for_user(user.id).await;
146
147 Ok(Json(BillingPreferencesResponse {
148 max_monthly_llm_usage_spending_in_cents: billing_preferences
149 .max_monthly_llm_usage_spending_in_cents,
150 }))
151}
152
153#[derive(Debug, Deserialize)]
154struct ListBillingSubscriptionsParams {
155 github_user_id: i32,
156}
157
158#[derive(Debug, Serialize)]
159struct BillingSubscriptionJson {
160 id: BillingSubscriptionId,
161 name: String,
162 status: StripeSubscriptionStatus,
163 cancel_at: Option<String>,
164 /// Whether this subscription can be canceled.
165 is_cancelable: bool,
166}
167
168#[derive(Debug, Serialize)]
169struct ListBillingSubscriptionsResponse {
170 subscriptions: Vec<BillingSubscriptionJson>,
171}
172
173async fn list_billing_subscriptions(
174 Extension(app): Extension<Arc<AppState>>,
175 Query(params): Query<ListBillingSubscriptionsParams>,
176) -> Result<Json<ListBillingSubscriptionsResponse>> {
177 let user = app
178 .db
179 .get_user_by_github_user_id(params.github_user_id)
180 .await?
181 .ok_or_else(|| anyhow!("user not found"))?;
182
183 let subscriptions = app.db.get_billing_subscriptions(user.id).await?;
184
185 Ok(Json(ListBillingSubscriptionsResponse {
186 subscriptions: subscriptions
187 .into_iter()
188 .map(|subscription| BillingSubscriptionJson {
189 id: subscription.id,
190 name: "Zed LLM Usage".to_string(),
191 status: subscription.stripe_subscription_status,
192 cancel_at: subscription.stripe_cancel_at.map(|cancel_at| {
193 cancel_at
194 .and_utc()
195 .to_rfc3339_opts(SecondsFormat::Millis, true)
196 }),
197 is_cancelable: subscription.stripe_subscription_status.is_cancelable()
198 && subscription.stripe_cancel_at.is_none(),
199 })
200 .collect(),
201 }))
202}
203
204#[derive(Debug, Deserialize)]
205struct CreateBillingSubscriptionBody {
206 github_user_id: i32,
207}
208
209#[derive(Debug, Serialize)]
210struct CreateBillingSubscriptionResponse {
211 checkout_session_url: String,
212}
213
214/// Initiates a Stripe Checkout session for creating a billing subscription.
215async fn create_billing_subscription(
216 Extension(app): Extension<Arc<AppState>>,
217 extract::Json(body): extract::Json<CreateBillingSubscriptionBody>,
218) -> Result<Json<CreateBillingSubscriptionResponse>> {
219 let user = app
220 .db
221 .get_user_by_github_user_id(body.github_user_id)
222 .await?
223 .ok_or_else(|| anyhow!("user not found"))?;
224
225 let Some(stripe_client) = app.stripe_client.clone() else {
226 log::error!("failed to retrieve Stripe client");
227 Err(Error::http(
228 StatusCode::NOT_IMPLEMENTED,
229 "not supported".into(),
230 ))?
231 };
232 let Some(stripe_billing) = app.stripe_billing.clone() else {
233 log::error!("failed to retrieve Stripe billing object");
234 Err(Error::http(
235 StatusCode::NOT_IMPLEMENTED,
236 "not supported".into(),
237 ))?
238 };
239 let Some(llm_db) = app.llm_db.clone() else {
240 log::error!("failed to retrieve LLM database");
241 Err(Error::http(
242 StatusCode::NOT_IMPLEMENTED,
243 "not supported".into(),
244 ))?
245 };
246
247 if app.db.has_active_billing_subscription(user.id).await? {
248 return Err(Error::http(
249 StatusCode::CONFLICT,
250 "user already has an active subscription".into(),
251 ));
252 }
253
254 let customer_id =
255 if let Some(existing_customer) = app.db.get_billing_customer_by_user_id(user.id).await? {
256 CustomerId::from_str(&existing_customer.stripe_customer_id)
257 .context("failed to parse customer ID")?
258 } else {
259 let customer = Customer::create(
260 &stripe_client,
261 CreateCustomer {
262 email: user.email_address.as_deref(),
263 ..Default::default()
264 },
265 )
266 .await?;
267
268 customer.id
269 };
270
271 let default_model = llm_db.model(rpc::LanguageModelProvider::Anthropic, "claude-3-5-sonnet")?;
272 let stripe_model = stripe_billing.register_model(default_model).await?;
273 let success_url = format!(
274 "{}/account?checkout_complete=1",
275 app.config.zed_dot_dev_url()
276 );
277 let checkout_session_url = stripe_billing
278 .checkout(customer_id, &user.github_login, &stripe_model, &success_url)
279 .await?;
280 Ok(Json(CreateBillingSubscriptionResponse {
281 checkout_session_url,
282 }))
283}
284
285#[derive(Debug, PartialEq, Deserialize)]
286#[serde(rename_all = "snake_case")]
287enum ManageSubscriptionIntent {
288 /// The user intends to cancel their subscription.
289 Cancel,
290 /// The user intends to stop the cancellation of their subscription.
291 StopCancellation,
292}
293
294#[derive(Debug, Deserialize)]
295struct ManageBillingSubscriptionBody {
296 github_user_id: i32,
297 intent: ManageSubscriptionIntent,
298 /// The ID of the subscription to manage.
299 subscription_id: BillingSubscriptionId,
300}
301
302#[derive(Debug, Serialize)]
303struct ManageBillingSubscriptionResponse {
304 billing_portal_session_url: Option<String>,
305}
306
307/// Initiates a Stripe customer portal session for managing a billing subscription.
308async fn manage_billing_subscription(
309 Extension(app): Extension<Arc<AppState>>,
310 extract::Json(body): extract::Json<ManageBillingSubscriptionBody>,
311) -> Result<Json<ManageBillingSubscriptionResponse>> {
312 let user = app
313 .db
314 .get_user_by_github_user_id(body.github_user_id)
315 .await?
316 .ok_or_else(|| anyhow!("user not found"))?;
317
318 let Some(stripe_client) = app.stripe_client.clone() else {
319 log::error!("failed to retrieve Stripe client");
320 Err(Error::http(
321 StatusCode::NOT_IMPLEMENTED,
322 "not supported".into(),
323 ))?
324 };
325
326 let customer = app
327 .db
328 .get_billing_customer_by_user_id(user.id)
329 .await?
330 .ok_or_else(|| anyhow!("billing customer not found"))?;
331 let customer_id = CustomerId::from_str(&customer.stripe_customer_id)
332 .context("failed to parse customer ID")?;
333
334 let subscription = app
335 .db
336 .get_billing_subscription_by_id(body.subscription_id)
337 .await?
338 .ok_or_else(|| anyhow!("subscription not found"))?;
339
340 if body.intent == ManageSubscriptionIntent::StopCancellation {
341 let subscription_id = SubscriptionId::from_str(&subscription.stripe_subscription_id)
342 .context("failed to parse subscription ID")?;
343
344 let updated_stripe_subscription = Subscription::update(
345 &stripe_client,
346 &subscription_id,
347 stripe::UpdateSubscription {
348 cancel_at_period_end: Some(false),
349 ..Default::default()
350 },
351 )
352 .await?;
353
354 app.db
355 .update_billing_subscription(
356 subscription.id,
357 &UpdateBillingSubscriptionParams {
358 stripe_cancel_at: ActiveValue::set(
359 updated_stripe_subscription
360 .cancel_at
361 .and_then(|cancel_at| DateTime::from_timestamp(cancel_at, 0))
362 .map(|time| time.naive_utc()),
363 ),
364 ..Default::default()
365 },
366 )
367 .await?;
368
369 return Ok(Json(ManageBillingSubscriptionResponse {
370 billing_portal_session_url: None,
371 }));
372 }
373
374 let flow = match body.intent {
375 ManageSubscriptionIntent::Cancel => CreateBillingPortalSessionFlowData {
376 type_: CreateBillingPortalSessionFlowDataType::SubscriptionCancel,
377 after_completion: Some(CreateBillingPortalSessionFlowDataAfterCompletion {
378 type_: stripe::CreateBillingPortalSessionFlowDataAfterCompletionType::Redirect,
379 redirect: Some(CreateBillingPortalSessionFlowDataAfterCompletionRedirect {
380 return_url: format!("{}/account", app.config.zed_dot_dev_url()),
381 }),
382 ..Default::default()
383 }),
384 subscription_cancel: Some(
385 stripe::CreateBillingPortalSessionFlowDataSubscriptionCancel {
386 subscription: subscription.stripe_subscription_id,
387 retention: None,
388 },
389 ),
390 ..Default::default()
391 },
392 ManageSubscriptionIntent::StopCancellation => unreachable!(),
393 };
394
395 let mut params = CreateBillingPortalSession::new(customer_id);
396 params.flow_data = Some(flow);
397 let return_url = format!("{}/account", app.config.zed_dot_dev_url());
398 params.return_url = Some(&return_url);
399
400 let session = BillingPortalSession::create(&stripe_client, params).await?;
401
402 Ok(Json(ManageBillingSubscriptionResponse {
403 billing_portal_session_url: Some(session.url),
404 }))
405}
406
407/// The amount of time we wait in between each poll of Stripe events.
408///
409/// This value should strike a balance between:
410/// 1. Being short enough that we update quickly when something in Stripe changes
411/// 2. Being long enough that we don't eat into our rate limits.
412///
413/// As a point of reference, the Sequin folks say they have this at **500ms**:
414///
415/// > We poll the Stripe /events endpoint every 500ms per account
416/// >
417/// > — https://blog.sequinstream.com/events-not-webhooks/
418const POLL_EVENTS_INTERVAL: Duration = Duration::from_secs(5);
419
420/// The maximum number of events to return per page.
421///
422/// We set this to 100 (the max) so we have to make fewer requests to Stripe.
423///
424/// > Limit can range between 1 and 100, and the default is 10.
425const EVENTS_LIMIT_PER_PAGE: u64 = 100;
426
427/// The number of pages consisting entirely of already-processed events that we
428/// will see before we stop retrieving events.
429///
430/// This is used to prevent over-fetching the Stripe events API for events we've
431/// already seen and processed.
432const NUMBER_OF_ALREADY_PROCESSED_PAGES_BEFORE_WE_STOP: usize = 4;
433
434/// Polls the Stripe events API periodically to reconcile the records in our
435/// database with the data in Stripe.
436pub fn poll_stripe_events_periodically(app: Arc<AppState>, rpc_server: Arc<Server>) {
437 let Some(stripe_client) = app.stripe_client.clone() else {
438 log::warn!("failed to retrieve Stripe client");
439 return;
440 };
441
442 let executor = app.executor.clone();
443 executor.spawn_detached({
444 let executor = executor.clone();
445 async move {
446 loop {
447 poll_stripe_events(&app, &rpc_server, &stripe_client)
448 .await
449 .log_err();
450
451 executor.sleep(POLL_EVENTS_INTERVAL).await;
452 }
453 }
454 });
455}
456
457async fn poll_stripe_events(
458 app: &Arc<AppState>,
459 rpc_server: &Arc<Server>,
460 stripe_client: &stripe::Client,
461) -> anyhow::Result<()> {
462 fn event_type_to_string(event_type: EventType) -> String {
463 // Calling `to_string` on `stripe::EventType` members gives us a quoted string,
464 // so we need to unquote it.
465 event_type.to_string().trim_matches('"').to_string()
466 }
467
468 let event_types = [
469 EventType::CustomerCreated,
470 EventType::CustomerUpdated,
471 EventType::CustomerSubscriptionCreated,
472 EventType::CustomerSubscriptionUpdated,
473 EventType::CustomerSubscriptionPaused,
474 EventType::CustomerSubscriptionResumed,
475 EventType::CustomerSubscriptionDeleted,
476 ]
477 .into_iter()
478 .map(event_type_to_string)
479 .collect::<Vec<_>>();
480
481 let mut pages_of_already_processed_events = 0;
482 let mut unprocessed_events = Vec::new();
483
484 log::info!(
485 "Stripe events: starting retrieval for {}",
486 event_types.join(", ")
487 );
488 let mut params = ListEvents::new();
489 params.types = Some(event_types.clone());
490 params.limit = Some(EVENTS_LIMIT_PER_PAGE);
491
492 let mut event_pages = stripe::Event::list(&stripe_client, ¶ms)
493 .await?
494 .paginate(params);
495
496 loop {
497 let processed_event_ids = {
498 let event_ids = event_pages
499 .page
500 .data
501 .iter()
502 .map(|event| event.id.as_str())
503 .collect::<Vec<_>>();
504 app.db
505 .get_processed_stripe_events_by_event_ids(&event_ids)
506 .await?
507 .into_iter()
508 .map(|event| event.stripe_event_id)
509 .collect::<Vec<_>>()
510 };
511
512 let mut processed_events_in_page = 0;
513 let events_in_page = event_pages.page.data.len();
514 for event in &event_pages.page.data {
515 if processed_event_ids.contains(&event.id.to_string()) {
516 processed_events_in_page += 1;
517 log::debug!("Stripe events: already processed '{}', skipping", event.id);
518 } else {
519 unprocessed_events.push(event.clone());
520 }
521 }
522
523 if processed_events_in_page == events_in_page {
524 pages_of_already_processed_events += 1;
525 }
526
527 if event_pages.page.has_more {
528 if pages_of_already_processed_events >= NUMBER_OF_ALREADY_PROCESSED_PAGES_BEFORE_WE_STOP
529 {
530 log::info!("Stripe events: stopping, saw {pages_of_already_processed_events} pages of already-processed events");
531 break;
532 } else {
533 log::info!("Stripe events: retrieving next page");
534 event_pages = event_pages.next(&stripe_client).await?;
535 }
536 } else {
537 break;
538 }
539 }
540
541 log::info!("Stripe events: unprocessed {}", unprocessed_events.len());
542
543 // Sort all of the unprocessed events in ascending order, so we can handle them in the order they occurred.
544 unprocessed_events.sort_by(|a, b| a.created.cmp(&b.created).then_with(|| a.id.cmp(&b.id)));
545
546 for event in unprocessed_events {
547 let event_id = event.id.clone();
548 let processed_event_params = CreateProcessedStripeEventParams {
549 stripe_event_id: event.id.to_string(),
550 stripe_event_type: event_type_to_string(event.type_),
551 stripe_event_created_timestamp: event.created,
552 };
553
554 // If the event has happened too far in the past, we don't want to
555 // process it and risk overwriting other more-recent updates.
556 //
557 // 1 day was chosen arbitrarily. This could be made longer or shorter.
558 let one_day = Duration::from_secs(24 * 60 * 60);
559 let a_day_ago = Utc::now() - one_day;
560 if a_day_ago.timestamp() > event.created {
561 log::info!(
562 "Stripe events: event '{}' is more than {one_day:?} old, marking as processed",
563 event_id
564 );
565 app.db
566 .create_processed_stripe_event(&processed_event_params)
567 .await?;
568
569 return Ok(());
570 }
571
572 let process_result = match event.type_ {
573 EventType::CustomerCreated | EventType::CustomerUpdated => {
574 handle_customer_event(app, stripe_client, event).await
575 }
576 EventType::CustomerSubscriptionCreated
577 | EventType::CustomerSubscriptionUpdated
578 | EventType::CustomerSubscriptionPaused
579 | EventType::CustomerSubscriptionResumed
580 | EventType::CustomerSubscriptionDeleted => {
581 handle_customer_subscription_event(app, rpc_server, stripe_client, event).await
582 }
583 _ => Ok(()),
584 };
585
586 if let Some(()) = process_result
587 .with_context(|| format!("failed to process event {event_id} successfully"))
588 .log_err()
589 {
590 app.db
591 .create_processed_stripe_event(&processed_event_params)
592 .await?;
593 }
594 }
595
596 Ok(())
597}
598
599async fn handle_customer_event(
600 app: &Arc<AppState>,
601 _stripe_client: &stripe::Client,
602 event: stripe::Event,
603) -> anyhow::Result<()> {
604 let EventObject::Customer(customer) = event.data.object else {
605 bail!("unexpected event payload for {}", event.id);
606 };
607
608 log::info!("handling Stripe {} event: {}", event.type_, event.id);
609
610 let Some(email) = customer.email else {
611 log::info!("Stripe customer has no email: skipping");
612 return Ok(());
613 };
614
615 let Some(user) = app.db.get_user_by_email(&email).await? else {
616 log::info!("no user found for email: skipping");
617 return Ok(());
618 };
619
620 if let Some(existing_customer) = app
621 .db
622 .get_billing_customer_by_stripe_customer_id(&customer.id)
623 .await?
624 {
625 app.db
626 .update_billing_customer(
627 existing_customer.id,
628 &UpdateBillingCustomerParams {
629 // For now we just leave the information as-is, as it is not
630 // likely to change.
631 ..Default::default()
632 },
633 )
634 .await?;
635 } else {
636 app.db
637 .create_billing_customer(&CreateBillingCustomerParams {
638 user_id: user.id,
639 stripe_customer_id: customer.id.to_string(),
640 })
641 .await?;
642 }
643
644 Ok(())
645}
646
647async fn handle_customer_subscription_event(
648 app: &Arc<AppState>,
649 rpc_server: &Arc<Server>,
650 stripe_client: &stripe::Client,
651 event: stripe::Event,
652) -> anyhow::Result<()> {
653 let EventObject::Subscription(subscription) = event.data.object else {
654 bail!("unexpected event payload for {}", event.id);
655 };
656
657 log::info!("handling Stripe {} event: {}", event.type_, event.id);
658
659 let billing_customer =
660 find_or_create_billing_customer(app, stripe_client, subscription.customer)
661 .await?
662 .ok_or_else(|| anyhow!("billing customer not found"))?;
663
664 if let Some(existing_subscription) = app
665 .db
666 .get_billing_subscription_by_stripe_subscription_id(&subscription.id)
667 .await?
668 {
669 app.db
670 .update_billing_subscription(
671 existing_subscription.id,
672 &UpdateBillingSubscriptionParams {
673 billing_customer_id: ActiveValue::set(billing_customer.id),
674 stripe_subscription_id: ActiveValue::set(subscription.id.to_string()),
675 stripe_subscription_status: ActiveValue::set(subscription.status.into()),
676 stripe_cancel_at: ActiveValue::set(
677 subscription
678 .cancel_at
679 .and_then(|cancel_at| DateTime::from_timestamp(cancel_at, 0))
680 .map(|time| time.naive_utc()),
681 ),
682 },
683 )
684 .await?;
685 } else {
686 // If the user already has an active billing subscription, ignore the
687 // event and return an `Ok` to signal that it was processed
688 // successfully.
689 //
690 // There is the possibility that this could cause us to not create a
691 // subscription in the following scenario:
692 //
693 // 1. User has an active subscription A
694 // 2. User cancels subscription A
695 // 3. User creates a new subscription B
696 // 4. We process the new subscription B before the cancellation of subscription A
697 // 5. User ends up with no subscriptions
698 //
699 // In theory this situation shouldn't arise as we try to process the events in the order they occur.
700 if app
701 .db
702 .has_active_billing_subscription(billing_customer.user_id)
703 .await?
704 {
705 log::info!(
706 "user {user_id} already has an active subscription, skipping creation of subscription {subscription_id}",
707 user_id = billing_customer.user_id,
708 subscription_id = subscription.id
709 );
710 return Ok(());
711 }
712
713 app.db
714 .create_billing_subscription(&CreateBillingSubscriptionParams {
715 billing_customer_id: billing_customer.id,
716 stripe_subscription_id: subscription.id.to_string(),
717 stripe_subscription_status: subscription.status.into(),
718 })
719 .await?;
720 }
721
722 // When the user's subscription changes, we want to refresh their LLM tokens
723 // to either grant/revoke access.
724 rpc_server
725 .refresh_llm_tokens_for_user(billing_customer.user_id)
726 .await;
727
728 Ok(())
729}
730
731#[derive(Debug, Deserialize)]
732struct GetMonthlySpendParams {
733 github_user_id: i32,
734}
735
736#[derive(Debug, Serialize)]
737struct GetMonthlySpendResponse {
738 monthly_free_tier_spend_in_cents: u32,
739 monthly_free_tier_allowance_in_cents: u32,
740 monthly_spend_in_cents: u32,
741}
742
743async fn get_monthly_spend(
744 Extension(app): Extension<Arc<AppState>>,
745 Query(params): Query<GetMonthlySpendParams>,
746) -> Result<Json<GetMonthlySpendResponse>> {
747 let user = app
748 .db
749 .get_user_by_github_user_id(params.github_user_id)
750 .await?
751 .ok_or_else(|| anyhow!("user not found"))?;
752
753 let Some(llm_db) = app.llm_db.clone() else {
754 return Err(Error::http(
755 StatusCode::NOT_IMPLEMENTED,
756 "LLM database not available".into(),
757 ));
758 };
759
760 let free_tier = user
761 .custom_llm_monthly_allowance_in_cents
762 .map(|allowance| Cents(allowance as u32))
763 .unwrap_or(FREE_TIER_MONTHLY_SPENDING_LIMIT);
764
765 let spending_for_month = llm_db
766 .get_user_spending_for_month(user.id, Utc::now())
767 .await?;
768
769 let free_tier_spend = Cents::min(spending_for_month, free_tier);
770 let monthly_spend = spending_for_month.saturating_sub(free_tier);
771
772 Ok(Json(GetMonthlySpendResponse {
773 monthly_free_tier_spend_in_cents: free_tier_spend.0,
774 monthly_free_tier_allowance_in_cents: free_tier.0,
775 monthly_spend_in_cents: monthly_spend.0,
776 }))
777}
778
779impl From<SubscriptionStatus> for StripeSubscriptionStatus {
780 fn from(value: SubscriptionStatus) -> Self {
781 match value {
782 SubscriptionStatus::Incomplete => Self::Incomplete,
783 SubscriptionStatus::IncompleteExpired => Self::IncompleteExpired,
784 SubscriptionStatus::Trialing => Self::Trialing,
785 SubscriptionStatus::Active => Self::Active,
786 SubscriptionStatus::PastDue => Self::PastDue,
787 SubscriptionStatus::Canceled => Self::Canceled,
788 SubscriptionStatus::Unpaid => Self::Unpaid,
789 SubscriptionStatus::Paused => Self::Paused,
790 }
791 }
792}
793
794/// Finds or creates a billing customer using the provided customer.
795async fn find_or_create_billing_customer(
796 app: &Arc<AppState>,
797 stripe_client: &stripe::Client,
798 customer_or_id: Expandable<Customer>,
799) -> anyhow::Result<Option<billing_customer::Model>> {
800 let customer_id = match &customer_or_id {
801 Expandable::Id(id) => id,
802 Expandable::Object(customer) => customer.id.as_ref(),
803 };
804
805 // If we already have a billing customer record associated with the Stripe customer,
806 // there's nothing more we need to do.
807 if let Some(billing_customer) = app
808 .db
809 .get_billing_customer_by_stripe_customer_id(customer_id)
810 .await?
811 {
812 return Ok(Some(billing_customer));
813 }
814
815 // If all we have is a customer ID, resolve it to a full customer record by
816 // hitting the Stripe API.
817 let customer = match customer_or_id {
818 Expandable::Id(id) => Customer::retrieve(stripe_client, &id, &[]).await?,
819 Expandable::Object(customer) => *customer,
820 };
821
822 let Some(email) = customer.email else {
823 return Ok(None);
824 };
825
826 let Some(user) = app.db.get_user_by_email(&email).await? else {
827 return Ok(None);
828 };
829
830 let billing_customer = app
831 .db
832 .create_billing_customer(&CreateBillingCustomerParams {
833 user_id: user.id,
834 stripe_customer_id: customer.id.to_string(),
835 })
836 .await?;
837
838 Ok(Some(billing_customer))
839}
840
841const SYNC_LLM_USAGE_WITH_STRIPE_INTERVAL: Duration = Duration::from_secs(60);
842
843pub fn sync_llm_usage_with_stripe_periodically(app: Arc<AppState>) {
844 let Some(stripe_billing) = app.stripe_billing.clone() else {
845 log::warn!("failed to retrieve Stripe billing object");
846 return;
847 };
848 let Some(llm_db) = app.llm_db.clone() else {
849 log::warn!("failed to retrieve LLM database");
850 return;
851 };
852
853 let executor = app.executor.clone();
854 executor.spawn_detached({
855 let executor = executor.clone();
856 async move {
857 loop {
858 sync_with_stripe(&app, &llm_db, &stripe_billing)
859 .await
860 .context("failed to sync LLM usage to Stripe")
861 .trace_err();
862 executor.sleep(SYNC_LLM_USAGE_WITH_STRIPE_INTERVAL).await;
863 }
864 }
865 });
866}
867
868async fn sync_with_stripe(
869 app: &Arc<AppState>,
870 llm_db: &Arc<LlmDatabase>,
871 stripe_billing: &Arc<StripeBilling>,
872) -> anyhow::Result<()> {
873 let events = llm_db.get_billing_events().await?;
874 let user_ids = events
875 .iter()
876 .map(|(event, _)| event.user_id)
877 .collect::<HashSet<UserId>>();
878 let stripe_subscriptions = app.db.get_active_billing_subscriptions(user_ids).await?;
879
880 for (event, model) in events {
881 let Some((stripe_db_customer, stripe_db_subscription)) =
882 stripe_subscriptions.get(&event.user_id)
883 else {
884 tracing::warn!(
885 user_id = event.user_id.0,
886 "Registered billing event for user who is not a Stripe customer. Billing events should only be created for users who are Stripe customers, so this is a mistake on our side."
887 );
888 continue;
889 };
890 let stripe_subscription_id: stripe::SubscriptionId = stripe_db_subscription
891 .stripe_subscription_id
892 .parse()
893 .context("failed to parse stripe subscription id from db")?;
894 let stripe_customer_id: stripe::CustomerId = stripe_db_customer
895 .stripe_customer_id
896 .parse()
897 .context("failed to parse stripe customer id from db")?;
898
899 let stripe_model = stripe_billing.register_model(&model).await?;
900 stripe_billing
901 .subscribe_to_model(&stripe_subscription_id, &stripe_model)
902 .await?;
903 stripe_billing
904 .bill_model_usage(&stripe_customer_id, &stripe_model, &event)
905 .await?;
906 llm_db.consume_billing_event(event.id).await?;
907 }
908
909 Ok(())
910}