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