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