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