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