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