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