use anyhow::{anyhow, bail, Context};use axum::{ extract::{self, Query}, routing::{get, post}, Extension, Json, Router,};use chrono::{DateTime, SecondsFormat, Utc};use collections::HashSet;use reqwest::StatusCode;use sea_orm::ActiveValue;use serde::{Deserialize, Serialize};use std::{str::FromStr, sync::Arc, time::Duration};use stripe::{ BillingPortalSession, CreateBillingPortalSession, CreateBillingPortalSessionFlowData, CreateBillingPortalSessionFlowDataAfterCompletion, CreateBillingPortalSessionFlowDataAfterCompletionRedirect, CreateBillingPortalSessionFlowDataType, CreateCustomer, Customer, CustomerId, EventObject, EventType, Expandable, ListEvents, Subscription, SubscriptionId, SubscriptionStatus,};use util::ResultExt; use crate::llm::{DEFAULT_MAX_MONTHLY_SPEND, FREE_TIER_MONTHLY_SPENDING_LIMIT};use crate::rpc::{ResultExt as _, Server};use crate::{ db::{ billing_customer, BillingSubscriptionId, CreateBillingCustomerParams, CreateBillingSubscriptionParams, CreateProcessedStripeEventParams, UpdateBillingCustomerParams, UpdateBillingPreferencesParams, UpdateBillingSubscriptionParams, }, stripe_billing::StripeBilling,};use crate::{ db::{billing_subscription::StripeSubscriptionStatus, UserId}, llm::db::LlmDatabase,};use crate::{AppState, Cents, Error, Result}; pub fn router() -> Router { Router::new() .route( "/billing/preferences", get(get_billing_preferences).put(update_billing_preferences), ) .route( "/billing/subscriptions", get(list_billing_subscriptions).post(create_billing_subscription), ) .route( "/billing/subscriptions/manage", post(manage_billing_subscription), ) .route("/billing/monthly_spend", get(get_monthly_spend))} #[derive(Debug, Deserialize)]struct GetBillingPreferencesParams { github_user_id: i32,} #[derive(Debug, Serialize)]struct BillingPreferencesResponse { max_monthly_llm_usage_spending_in_cents: i32,} async fn get_billing_preferences( Extension(app): Extension<Arc<AppState>>, Query(params): Query<GetBillingPreferencesParams>,) -> Result<Json<BillingPreferencesResponse>> { let user = app .db .get_user_by_github_user_id(params.github_user_id) .await? .ok_or_else(|| anyhow!("user not found"))?; let preferences = app.db.get_billing_preferences(user.id).await?; Ok(Json(BillingPreferencesResponse { max_monthly_llm_usage_spending_in_cents: preferences .map_or(DEFAULT_MAX_MONTHLY_SPEND.0 as i32, |preferences| { preferences.max_monthly_llm_usage_spending_in_cents }), }))} #[derive(Debug, Deserialize)]struct UpdateBillingPreferencesBody { github_user_id: i32, max_monthly_llm_usage_spending_in_cents: i32,} async fn update_billing_preferences( Extension(app): Extension<Arc<AppState>>, Extension(rpc_server): Extension<Arc<crate::rpc::Server>>, extract::Json(body): extract::Json<UpdateBillingPreferencesBody>,) -> Result<Json<BillingPreferencesResponse>> { let user = app .db .get_user_by_github_user_id(body.github_user_id) .await? .ok_or_else(|| anyhow!("user not found"))?; let billing_preferences = if let Some(_billing_preferences) = app.db.get_billing_preferences(user.id).await? { app.db .update_billing_preferences( user.id, &UpdateBillingPreferencesParams { max_monthly_llm_usage_spending_in_cents: ActiveValue::set( body.max_monthly_llm_usage_spending_in_cents, ), }, ) .await? } else { app.db .create_billing_preferences( user.id, &crate::db::CreateBillingPreferencesParams { max_monthly_llm_usage_spending_in_cents: body .max_monthly_llm_usage_spending_in_cents, }, ) .await? }; rpc_server.refresh_llm_tokens_for_user(user.id).await; Ok(Json(BillingPreferencesResponse { max_monthly_llm_usage_spending_in_cents: billing_preferences .max_monthly_llm_usage_spending_in_cents, }))} #[derive(Debug, Deserialize)]struct ListBillingSubscriptionsParams { github_user_id: i32,} #[derive(Debug, Serialize)]struct BillingSubscriptionJson { id: BillingSubscriptionId, name: String, status: StripeSubscriptionStatus, cancel_at: Option<String>, /// Whether this subscription can be canceled. is_cancelable: bool,} #[derive(Debug, Serialize)]struct ListBillingSubscriptionsResponse { subscriptions: Vec<BillingSubscriptionJson>,} async fn list_billing_subscriptions( Extension(app): Extension<Arc<AppState>>, Query(params): Query<ListBillingSubscriptionsParams>,) -> Result<Json<ListBillingSubscriptionsResponse>> { let user = app .db .get_user_by_github_user_id(params.github_user_id) .await? .ok_or_else(|| anyhow!("user not found"))?; let subscriptions = app.db.get_billing_subscriptions(user.id).await?; Ok(Json(ListBillingSubscriptionsResponse { subscriptions: subscriptions .into_iter() .map(|subscription| BillingSubscriptionJson { id: subscription.id, name: "Zed LLM Usage".to_string(), status: subscription.stripe_subscription_status, cancel_at: subscription.stripe_cancel_at.map(|cancel_at| { cancel_at .and_utc() .to_rfc3339_opts(SecondsFormat::Millis, true) }), is_cancelable: subscription.stripe_subscription_status.is_cancelable() && subscription.stripe_cancel_at.is_none(), }) .collect(), }))} #[derive(Debug, Deserialize)]struct CreateBillingSubscriptionBody { github_user_id: i32,} #[derive(Debug, Serialize)]struct CreateBillingSubscriptionResponse { checkout_session_url: String,} /// Initiates a Stripe Checkout session for creating a billing subscription.async fn create_billing_subscription( Extension(app): Extension<Arc<AppState>>, extract::Json(body): extract::Json<CreateBillingSubscriptionBody>,) -> Result<Json<CreateBillingSubscriptionResponse>> { let user = app .db .get_user_by_github_user_id(body.github_user_id) .await? .ok_or_else(|| anyhow!("user not found"))?; let Some(stripe_client) = app.stripe_client.clone() else { log::error!("failed to retrieve Stripe client"); Err(Error::http( StatusCode::NOT_IMPLEMENTED, "not supported".into(), ))? }; let Some(stripe_billing) = app.stripe_billing.clone() else { log::error!("failed to retrieve Stripe billing object"); Err(Error::http( StatusCode::NOT_IMPLEMENTED, "not supported".into(), ))? }; let Some(llm_db) = app.llm_db.clone() else { log::error!("failed to retrieve LLM database"); Err(Error::http( StatusCode::NOT_IMPLEMENTED, "not supported".into(), ))? }; if app.db.has_active_billing_subscription(user.id).await? { return Err(Error::http( StatusCode::CONFLICT, "user already has an active subscription".into(), )); } let customer_id = if let Some(existing_customer) = app.db.get_billing_customer_by_user_id(user.id).await? { CustomerId::from_str(&existing_customer.stripe_customer_id) .context("failed to parse customer ID")? } else { let customer = Customer::create( &stripe_client, CreateCustomer { email: user.email_address.as_deref(), ..Default::default() }, ) .await?; customer.id }; let default_model = llm_db.model(rpc::LanguageModelProvider::Anthropic, "claude-3-5-sonnet")?; let stripe_model = stripe_billing.register_model(default_model).await?; let success_url = format!("{}/account", app.config.zed_dot_dev_url()); let checkout_session_url = stripe_billing .checkout(customer_id, &user.github_login, &stripe_model, &success_url) .await?; Ok(Json(CreateBillingSubscriptionResponse { checkout_session_url, }))} #[derive(Debug, PartialEq, Deserialize)]#[serde(rename_all = "snake_case")]enum ManageSubscriptionIntent { /// The user intends to cancel their subscription. Cancel, /// The user intends to stop the cancellation of their subscription. StopCancellation,} #[derive(Debug, Deserialize)]struct ManageBillingSubscriptionBody { github_user_id: i32, intent: ManageSubscriptionIntent, /// The ID of the subscription to manage. subscription_id: BillingSubscriptionId,} #[derive(Debug, Serialize)]struct ManageBillingSubscriptionResponse { billing_portal_session_url: Option<String>,} /// Initiates a Stripe customer portal session for managing a billing subscription.async fn manage_billing_subscription( Extension(app): Extension<Arc<AppState>>, extract::Json(body): extract::Json<ManageBillingSubscriptionBody>,) -> Result<Json<ManageBillingSubscriptionResponse>> { let user = app .db .get_user_by_github_user_id(body.github_user_id) .await? .ok_or_else(|| anyhow!("user not found"))?; let Some(stripe_client) = app.stripe_client.clone() else { log::error!("failed to retrieve Stripe client"); Err(Error::http( StatusCode::NOT_IMPLEMENTED, "not supported".into(), ))? }; let customer = app .db .get_billing_customer_by_user_id(user.id) .await? .ok_or_else(|| anyhow!("billing customer not found"))?; let customer_id = CustomerId::from_str(&customer.stripe_customer_id) .context("failed to parse customer ID")?; let subscription = app .db .get_billing_subscription_by_id(body.subscription_id) .await? .ok_or_else(|| anyhow!("subscription not found"))?; if body.intent == ManageSubscriptionIntent::StopCancellation { let subscription_id = SubscriptionId::from_str(&subscription.stripe_subscription_id) .context("failed to parse subscription ID")?; let updated_stripe_subscription = Subscription::update( &stripe_client, &subscription_id, stripe::UpdateSubscription { cancel_at_period_end: Some(false), ..Default::default() }, ) .await?; app.db .update_billing_subscription( subscription.id, &UpdateBillingSubscriptionParams { stripe_cancel_at: ActiveValue::set( updated_stripe_subscription .cancel_at .and_then(|cancel_at| DateTime::from_timestamp(cancel_at, 0)) .map(|time| time.naive_utc()), ), ..Default::default() }, ) .await?; return Ok(Json(ManageBillingSubscriptionResponse { billing_portal_session_url: None, })); } let flow = match body.intent { ManageSubscriptionIntent::Cancel => CreateBillingPortalSessionFlowData { type_: CreateBillingPortalSessionFlowDataType::SubscriptionCancel, after_completion: Some(CreateBillingPortalSessionFlowDataAfterCompletion { type_: stripe::CreateBillingPortalSessionFlowDataAfterCompletionType::Redirect, redirect: Some(CreateBillingPortalSessionFlowDataAfterCompletionRedirect { return_url: format!("{}/account", app.config.zed_dot_dev_url()), }), ..Default::default() }), subscription_cancel: Some( stripe::CreateBillingPortalSessionFlowDataSubscriptionCancel { subscription: subscription.stripe_subscription_id, retention: None, }, ), ..Default::default() }, ManageSubscriptionIntent::StopCancellation => unreachable!(), }; let mut params = CreateBillingPortalSession::new(customer_id); params.flow_data = Some(flow); let return_url = format!("{}/account", app.config.zed_dot_dev_url()); params.return_url = Some(&return_url); let session = BillingPortalSession::create(&stripe_client, params).await?; Ok(Json(ManageBillingSubscriptionResponse { billing_portal_session_url: Some(session.url), }))} /// The amount of time we wait in between each poll of Stripe events.////// This value should strike a balance between:/// 1. Being short enough that we update quickly when something in Stripe changes/// 2. Being long enough that we don't eat into our rate limits.////// As a point of reference, the Sequin folks say they have this at **500ms**:////// > We poll the Stripe /events endpoint every 500ms per account/// >/// > — https://blog.sequinstream.com/events-not-webhooks/const POLL_EVENTS_INTERVAL: Duration = Duration::from_secs(5); /// The maximum number of events to return per page.////// We set this to 100 (the max) so we have to make fewer requests to Stripe.////// > Limit can range between 1 and 100, and the default is 10.const EVENTS_LIMIT_PER_PAGE: u64 = 100; /// The number of pages consisting entirely of already-processed events that we/// will see before we stop retrieving events.////// This is used to prevent over-fetching the Stripe events API for events we've/// already seen and processed.const NUMBER_OF_ALREADY_PROCESSED_PAGES_BEFORE_WE_STOP: usize = 4; /// Polls the Stripe events API periodically to reconcile the records in our/// database with the data in Stripe.pub fn poll_stripe_events_periodically(app: Arc<AppState>, rpc_server: Arc<Server>) { let Some(stripe_client) = app.stripe_client.clone() else { log::warn!("failed to retrieve Stripe client"); return; }; let executor = app.executor.clone(); executor.spawn_detached({ let executor = executor.clone(); async move { loop { poll_stripe_events(&app, &rpc_server, &stripe_client) .await .log_err(); executor.sleep(POLL_EVENTS_INTERVAL).await; } } });} async fn poll_stripe_events( app: &Arc<AppState>, rpc_server: &Arc<Server>, stripe_client: &stripe::Client,) -> anyhow::Result<()> { fn event_type_to_string(event_type: EventType) -> String { // Calling `to_string` on `stripe::EventType` members gives us a quoted string, // so we need to unquote it. event_type.to_string().trim_matches('"').to_string() } let event_types = [ EventType::CustomerCreated, EventType::CustomerUpdated, EventType::CustomerSubscriptionCreated, EventType::CustomerSubscriptionUpdated, EventType::CustomerSubscriptionPaused, EventType::CustomerSubscriptionResumed, EventType::CustomerSubscriptionDeleted, ] .into_iter() .map(event_type_to_string) .collect::<Vec<_>>(); let mut pages_of_already_processed_events = 0; let mut unprocessed_events = Vec::new(); log::info!( "Stripe events: starting retrieval for {}", event_types.join(", ") ); let mut params = ListEvents::new(); params.types = Some(event_types.clone()); params.limit = Some(EVENTS_LIMIT_PER_PAGE); let mut event_pages = stripe::Event::list(&stripe_client, ¶ms) .await? .paginate(params); loop { let processed_event_ids = { let event_ids = event_pages .page .data .iter() .map(|event| event.id.as_str()) .collect::<Vec<_>>(); app.db .get_processed_stripe_events_by_event_ids(&event_ids) .await? .into_iter() .map(|event| event.stripe_event_id) .collect::<Vec<_>>() }; let mut processed_events_in_page = 0; let events_in_page = event_pages.page.data.len(); for event in &event_pages.page.data { if processed_event_ids.contains(&event.id.to_string()) { processed_events_in_page += 1; log::debug!("Stripe events: already processed '{}', skipping", event.id); } else { unprocessed_events.push(event.clone()); } }Lines 1–500 of 889. Load more
use anyhow::{anyhow, bail, Context}; use axum::{ extract::{self, Query}, routing::{get, post}, Extension, Json, Router, }; use chrono::{DateTime, SecondsFormat, Utc}; use collections::HashSet; use reqwest::StatusCode; use sea_orm::ActiveValue; use serde::{Deserialize, Serialize}; use std::{str::FromStr, sync::Arc, time::Duration}; use stripe::{ BillingPortalSession, CreateBillingPortalSession, CreateBillingPortalSessionFlowData, CreateBillingPortalSessionFlowDataAfterCompletion, CreateBillingPortalSessionFlowDataAfterCompletionRedirect, CreateBillingPortalSessionFlowDataType, CreateCustomer, Customer, CustomerId, EventObject, EventType, Expandable, ListEvents, Subscription, SubscriptionId, SubscriptionStatus, }; use util::ResultExt; use crate::llm::{DEFAULT_MAX_MONTHLY_SPEND, FREE_TIER_MONTHLY_SPENDING_LIMIT}; use crate::rpc::{ResultExt as _, Server}; use crate::{ db::{ billing_customer, BillingSubscriptionId, CreateBillingCustomerParams, CreateBillingSubscriptionParams, CreateProcessedStripeEventParams, UpdateBillingCustomerParams, UpdateBillingPreferencesParams, UpdateBillingSubscriptionParams, }, stripe_billing::StripeBilling, }; use crate::{ db::{billing_subscription::StripeSubscriptionStatus, UserId}, llm::db::LlmDatabase, }; use crate::{AppState, Cents, Error, Result}; pub fn router() -> Router { Router::new() .route( "/billing/preferences", get(get_billing_preferences).put(update_billing_preferences), ) .route( "/billing/subscriptions", get(list_billing_subscriptions).post(create_billing_subscription), ) .route( "/billing/subscriptions/manage", post(manage_billing_subscription), ) .route("/billing/monthly_spend", get(get_monthly_spend)) } #[derive(Debug, Deserialize)] struct GetBillingPreferencesParams { github_user_id: i32, } #[derive(Debug, Serialize)] struct BillingPreferencesResponse { max_monthly_llm_usage_spending_in_cents: i32, } async fn get_billing_preferences( Extension(app): Extension<Arc<AppState>>, Query(params): Query<GetBillingPreferencesParams>, ) -> Result<Json<BillingPreferencesResponse>> { let user = app .db .get_user_by_github_user_id(params.github_user_id) .await? .ok_or_else(|| anyhow!("user not found"))?; let preferences = app.db.get_billing_preferences(user.id).await?; Ok(Json(BillingPreferencesResponse { max_monthly_llm_usage_spending_in_cents: preferences .map_or(DEFAULT_MAX_MONTHLY_SPEND.0 as i32, |preferences| { preferences.max_monthly_llm_usage_spending_in_cents }), })) } #[derive(Debug, Deserialize)] struct UpdateBillingPreferencesBody { github_user_id: i32, max_monthly_llm_usage_spending_in_cents: i32, } async fn update_billing_preferences( Extension(app): Extension<Arc<AppState>>, Extension(rpc_server): Extension<Arc<crate::rpc::Server>>, extract::Json(body): extract::Json<UpdateBillingPreferencesBody>, ) -> Result<Json<BillingPreferencesResponse>> { let user = app .db .get_user_by_github_user_id(body.github_user_id) .await? .ok_or_else(|| anyhow!("user not found"))?; let billing_preferences = if let Some(_billing_preferences) = app.db.get_billing_preferences(user.id).await? { app.db .update_billing_preferences( user.id, &UpdateBillingPreferencesParams { max_monthly_llm_usage_spending_in_cents: ActiveValue::set( body.max_monthly_llm_usage_spending_in_cents, ), }, ) .await? } else { app.db .create_billing_preferences( user.id, &crate::db::CreateBillingPreferencesParams { max_monthly_llm_usage_spending_in_cents: body .max_monthly_llm_usage_spending_in_cents, }, ) .await? }; rpc_server.refresh_llm_tokens_for_user(user.id).await; Ok(Json(BillingPreferencesResponse { max_monthly_llm_usage_spending_in_cents: billing_preferences .max_monthly_llm_usage_spending_in_cents, })) } #[derive(Debug, Deserialize)] struct ListBillingSubscriptionsParams { github_user_id: i32, } #[derive(Debug, Serialize)] struct BillingSubscriptionJson { id: BillingSubscriptionId, name: String, status: StripeSubscriptionStatus, cancel_at: Option<String>, /// Whether this subscription can be canceled. is_cancelable: bool, } #[derive(Debug, Serialize)] struct ListBillingSubscriptionsResponse { subscriptions: Vec<BillingSubscriptionJson>, } async fn list_billing_subscriptions( Extension(app): Extension<Arc<AppState>>, Query(params): Query<ListBillingSubscriptionsParams>, ) -> Result<Json<ListBillingSubscriptionsResponse>> { let user = app .db .get_user_by_github_user_id(params.github_user_id) .await? .ok_or_else(|| anyhow!("user not found"))?; let subscriptions = app.db.get_billing_subscriptions(user.id).await?; Ok(Json(ListBillingSubscriptionsResponse { subscriptions: subscriptions .into_iter() .map(|subscription| BillingSubscriptionJson { id: subscription.id, name: "Zed LLM Usage".to_string(), status: subscription.stripe_subscription_status, cancel_at: subscription.stripe_cancel_at.map(|cancel_at| { cancel_at .and_utc() .to_rfc3339_opts(SecondsFormat::Millis, true) }), is_cancelable: subscription.stripe_subscription_status.is_cancelable() && subscription.stripe_cancel_at.is_none(), }) .collect(), })) } #[derive(Debug, Deserialize)] struct CreateBillingSubscriptionBody { github_user_id: i32, } #[derive(Debug, Serialize)] struct CreateBillingSubscriptionResponse { checkout_session_url: String, } /// Initiates a Stripe Checkout session for creating a billing subscription. async fn create_billing_subscription( Extension(app): Extension<Arc<AppState>>, extract::Json(body): extract::Json<CreateBillingSubscriptionBody>, ) -> Result<Json<CreateBillingSubscriptionResponse>> { let user = app .db .get_user_by_github_user_id(body.github_user_id) .await? .ok_or_else(|| anyhow!("user not found"))?; let Some(stripe_client) = app.stripe_client.clone() else { log::error!("failed to retrieve Stripe client"); Err(Error::http( StatusCode::NOT_IMPLEMENTED, "not supported".into(), ))? }; let Some(stripe_billing) = app.stripe_billing.clone() else { log::error!("failed to retrieve Stripe billing object"); Err(Error::http( StatusCode::NOT_IMPLEMENTED, "not supported".into(), ))? }; let Some(llm_db) = app.llm_db.clone() else { log::error!("failed to retrieve LLM database"); Err(Error::http( StatusCode::NOT_IMPLEMENTED, "not supported".into(), ))? }; if app.db.has_active_billing_subscription(user.id).await? { return Err(Error::http( StatusCode::CONFLICT, "user already has an active subscription".into(), )); } let customer_id = if let Some(existing_customer) = app.db.get_billing_customer_by_user_id(user.id).await? { CustomerId::from_str(&existing_customer.stripe_customer_id) .context("failed to parse customer ID")? } else { let customer = Customer::create( &stripe_client, CreateCustomer { email: user.email_address.as_deref(), ..Default::default() }, ) .await?; customer.id }; let default_model = llm_db.model(rpc::LanguageModelProvider::Anthropic, "claude-3-5-sonnet")?; let stripe_model = stripe_billing.register_model(default_model).await?; let success_url = format!("{}/account", app.config.zed_dot_dev_url()); let checkout_session_url = stripe_billing .checkout(customer_id, &user.github_login, &stripe_model, &success_url) .await?; Ok(Json(CreateBillingSubscriptionResponse { checkout_session_url, })) } #[derive(Debug, PartialEq, Deserialize)] #[serde(rename_all = "snake_case")] enum ManageSubscriptionIntent { /// The user intends to cancel their subscription. Cancel, /// The user intends to stop the cancellation of their subscription. StopCancellation, } #[derive(Debug, Deserialize)] struct ManageBillingSubscriptionBody { github_user_id: i32, intent: ManageSubscriptionIntent, /// The ID of the subscription to manage. subscription_id: BillingSubscriptionId, } #[derive(Debug, Serialize)] struct ManageBillingSubscriptionResponse { billing_portal_session_url: Option<String>, } /// Initiates a Stripe customer portal session for managing a billing subscription. async fn manage_billing_subscription( Extension(app): Extension<Arc<AppState>>, extract::Json(body): extract::Json<ManageBillingSubscriptionBody>, ) -> Result<Json<ManageBillingSubscriptionResponse>> { let user = app .db .get_user_by_github_user_id(body.github_user_id) .await? .ok_or_else(|| anyhow!("user not found"))?; let Some(stripe_client) = app.stripe_client.clone() else { log::error!("failed to retrieve Stripe client"); Err(Error::http( StatusCode::NOT_IMPLEMENTED, "not supported".into(), ))? }; let customer = app .db .get_billing_customer_by_user_id(user.id) .await? .ok_or_else(|| anyhow!("billing customer not found"))?; let customer_id = CustomerId::from_str(&customer.stripe_customer_id) .context("failed to parse customer ID")?; let subscription = app .db .get_billing_subscription_by_id(body.subscription_id) .await? .ok_or_else(|| anyhow!("subscription not found"))?; if body.intent == ManageSubscriptionIntent::StopCancellation { let subscription_id = SubscriptionId::from_str(&subscription.stripe_subscription_id) .context("failed to parse subscription ID")?; let updated_stripe_subscription = Subscription::update( &stripe_client, &subscription_id, stripe::UpdateSubscription { cancel_at_period_end: Some(false), ..Default::default() }, ) .await?; app.db .update_billing_subscription( subscription.id, &UpdateBillingSubscriptionParams { stripe_cancel_at: ActiveValue::set( updated_stripe_subscription .cancel_at .and_then(|cancel_at| DateTime::from_timestamp(cancel_at, 0)) .map(|time| time.naive_utc()), ), ..Default::default() }, ) .await?; return Ok(Json(ManageBillingSubscriptionResponse { billing_portal_session_url: None, })); } let flow = match body.intent { ManageSubscriptionIntent::Cancel => CreateBillingPortalSessionFlowData { type_: CreateBillingPortalSessionFlowDataType::SubscriptionCancel, after_completion: Some(CreateBillingPortalSessionFlowDataAfterCompletion { type_: stripe::CreateBillingPortalSessionFlowDataAfterCompletionType::Redirect, redirect: Some(CreateBillingPortalSessionFlowDataAfterCompletionRedirect { return_url: format!("{}/account", app.config.zed_dot_dev_url()), }), ..Default::default() }), subscription_cancel: Some( stripe::CreateBillingPortalSessionFlowDataSubscriptionCancel { subscription: subscription.stripe_subscription_id, retention: None, }, ), ..Default::default() }, ManageSubscriptionIntent::StopCancellation => unreachable!(), }; let mut params = CreateBillingPortalSession::new(customer_id); params.flow_data = Some(flow); let return_url = format!("{}/account", app.config.zed_dot_dev_url()); params.return_url = Some(&return_url); let session = BillingPortalSession::create(&stripe_client, params).await?; Ok(Json(ManageBillingSubscriptionResponse { billing_portal_session_url: Some(session.url), })) } /// The amount of time we wait in between each poll of Stripe events. /// /// This value should strike a balance between: /// 1. Being short enough that we update quickly when something in Stripe changes /// 2. Being long enough that we don't eat into our rate limits. /// /// As a point of reference, the Sequin folks say they have this at **500ms**: /// /// > We poll the Stripe /events endpoint every 500ms per account /// > /// > — https://blog.sequinstream.com/events-not-webhooks/ const POLL_EVENTS_INTERVAL: Duration = Duration::from_secs(5); /// The maximum number of events to return per page. /// /// We set this to 100 (the max) so we have to make fewer requests to Stripe. /// /// > Limit can range between 1 and 100, and the default is 10. const EVENTS_LIMIT_PER_PAGE: u64 = 100; /// The number of pages consisting entirely of already-processed events that we /// will see before we stop retrieving events. /// /// This is used to prevent over-fetching the Stripe events API for events we've /// already seen and processed. const NUMBER_OF_ALREADY_PROCESSED_PAGES_BEFORE_WE_STOP: usize = 4; /// Polls the Stripe events API periodically to reconcile the records in our /// database with the data in Stripe. pub fn poll_stripe_events_periodically(app: Arc<AppState>, rpc_server: Arc<Server>) { let Some(stripe_client) = app.stripe_client.clone() else { log::warn!("failed to retrieve Stripe client"); return; }; let executor = app.executor.clone(); executor.spawn_detached({ let executor = executor.clone(); async move { loop { poll_stripe_events(&app, &rpc_server, &stripe_client) .await .log_err(); executor.sleep(POLL_EVENTS_INTERVAL).await; } } }); } async fn poll_stripe_events( app: &Arc<AppState>, rpc_server: &Arc<Server>, stripe_client: &stripe::Client, ) -> anyhow::Result<()> { fn event_type_to_string(event_type: EventType) -> String { // Calling `to_string` on `stripe::EventType` members gives us a quoted string, // so we need to unquote it. event_type.to_string().trim_matches('"').to_string() } let event_types = [ EventType::CustomerCreated, EventType::CustomerUpdated, EventType::CustomerSubscriptionCreated, EventType::CustomerSubscriptionUpdated, EventType::CustomerSubscriptionPaused, EventType::CustomerSubscriptionResumed, EventType::CustomerSubscriptionDeleted, ] .into_iter() .map(event_type_to_string) .collect::<Vec<_>>(); let mut pages_of_already_processed_events = 0; let mut unprocessed_events = Vec::new(); log::info!( "Stripe events: starting retrieval for {}", event_types.join(", ") ); let mut params = ListEvents::new(); params.types = Some(event_types.clone()); params.limit = Some(EVENTS_LIMIT_PER_PAGE); let mut event_pages = stripe::Event::list(&stripe_client, ¶ms) .await? .paginate(params); loop { let processed_event_ids = { let event_ids = event_pages .page .data .iter() .map(|event| event.id.as_str()) .collect::<Vec<_>>(); app.db .get_processed_stripe_events_by_event_ids(&event_ids) .await? .into_iter() .map(|event| event.stripe_event_id) .collect::<Vec<_>>() }; let mut processed_events_in_page = 0; let events_in_page = event_pages.page.data.len(); for event in &event_pages.page.data { if processed_event_ids.contains(&event.id.to_string()) { processed_events_in_page += 1; log::debug!("Stripe events: already processed '{}', skipping", event.id); } else { unprocessed_events.push(event.clone()); } }