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