autumn_billing/
lib.rs

1//! # autumn-billing
2//!
3//! Subscription billing for `autumn-web` applications: hosted checkout, the
4//! customer portal, a webhook-fed local mirror of billing state, a plan gate,
5//! and durable dunning. Stripe ships first behind a provider-neutral
6//! [`BillingProvider`] trait.
7//!
8//! # Quick start
9//!
10//! ```rust,ignore
11//! use autumn_billing::prelude::*;
12//!
13//! let billing = BillingConfig::from_autumn_toml("autumn.toml")?;
14//! let plans = PlanCatalog::new().plan(
15//!     Plan::new("pro", "Pro", "price_123", Money::from_minor(1999, Currency::USD), BillingInterval::Month)
16//!         .entitlement("export"),
17//! );
18//! autumn_web::app()
19//!     .plugin(BillingPlugin::new().config(billing).plans(&plans))
20//!     .run()
21//!     .await;
22//! ```
23//!
24//! Declare the webhook receiver in `autumn.toml` so `SignedWebhook` verifies
25//! it and CSRF exempts it. `max_body_bytes` is required: the webhook default is
26//! 1 MiB, a large provider event can exceed it, and boot fails below 4 MiB:
27//!
28//! ```toml
29//! [[security.webhooks.endpoints]]
30//! name = "billing"
31//! path = "/billing/webhook"
32//! provider = "stripe"
33//! secret_env = "STRIPE_WEBHOOK_SECRET"
34//! max_body_bytes = 4194304
35//! ```
36//!
37//! Gate a route:
38//!
39//! ```rust,ignore
40//! struct Pro;
41//! impl PlanRequirement for Pro {
42//!     fn rule() -> PlanRule { PlanRule::plan("pro") }
43//! }
44//!
45//! #[get("/reports")]
46//! async fn reports(_pro: Entitled<Pro>) -> &'static str { "ok" }
47//! ```
48
49pub mod config;
50pub mod dunning;
51pub mod error;
52pub mod event;
53pub mod gate;
54pub mod hooks;
55pub mod model;
56pub mod money;
57pub mod notify;
58pub mod plan;
59pub mod provider;
60pub mod reconcile;
61pub mod routes;
62pub mod store;
63mod stripe;
64
65pub use config::{BillingConfig, DunningPolicy, ExhaustionAction, SecretString, StripeConfig};
66pub use error::BillingError;
67pub use event::{
68    BillingEvent, BillingEventKind, CheckoutSnapshot, InvoiceSnapshot, SubscriptionSnapshot,
69};
70pub use gate::{Billing, Entitled, PlanRequirement, PlanRule, SubscriptionView};
71pub use hooks::{BillingHooks, NoHooks};
72pub use model::{
73    Customer, DunningAttempt, DunningState, Invoice, InvoiceStatus, ProviderId, Subscription,
74    SubscriptionStatus,
75};
76pub use money::{Currency, Money, MoneyError};
77pub use plan::{BillingInterval, Plan, PlanCatalog, PlanId};
78pub use provider::{
79    BillingProvider, CheckoutRequest, CustomerRequest, HostedSession, PaymentAttemptOutcome,
80    PortalRequest,
81};
82pub use reconcile::ReconcileOutcome;
83#[cfg(feature = "db")]
84pub use store::DbBillingStore;
85pub use store::{BillingStore, MemoryBillingStore};
86pub use stripe::StripeProvider;
87
88/// Common imports for configuring and mounting the plugin.
89pub mod prelude {
90    pub use crate::{
91        Billing, BillingConfig, BillingError, BillingHooks, BillingInterval, BillingPlugin,
92        BillingProvider, BillingStore, Currency, DunningPolicy, Entitled, ExhaustionAction,
93        MemoryBillingStore, Money, Plan, PlanCatalog, PlanId, PlanRequirement, PlanRule,
94        Subscription, SubscriptionStatus, SubscriptionView,
95    };
96}
97
98use std::borrow::Cow;
99use std::sync::Arc;
100
101use autumn_web::app::AppBuilder;
102use autumn_web::plugin::Plugin;
103use autumn_web::plugin_contract::PluginContract;
104use autumn_web::{AppState, AutumnError};
105
106/// How long a ledger claim in `processing` blocks redelivery before it is
107/// treated as abandoned.
108pub const EVENT_CLAIM_STALE_AFTER: std::time::Duration = std::time::Duration::from_secs(300);
109
110/// Embedded migrations for the mirror tables. Registered by [`BillingPlugin`].
111#[cfg(feature = "db")]
112pub const MIGRATIONS: autumn_web::migrate::EmbeddedMigrations = {
113    use autumn_web::reexports::diesel_migrations;
114    autumn_web::migrate::embed_migrations!("migrations")
115};
116
117/// Runtime handle installed on `AppState` by the plugin.
118pub struct BillingService {
119    provider: Arc<dyn BillingProvider>,
120    store: Arc<dyn BillingStore>,
121    catalog: Arc<PlanCatalog>,
122    config: Arc<BillingConfig>,
123    hooks: Arc<dyn BillingHooks>,
124    /// Users with a checkout request in progress.
125    checkouts_in_flight: std::sync::Mutex<std::collections::HashSet<String>>,
126}
127
128/// Holds a user's checkout slot. Dropping it frees the slot.
129pub(crate) struct CheckoutSlot {
130    service: Arc<BillingService>,
131    user_id: String,
132}
133
134impl Drop for CheckoutSlot {
135    fn drop(&mut self) {
136        self.service
137            .checkouts_in_flight
138            .lock()
139            .unwrap_or_else(std::sync::PoisonError::into_inner)
140            .remove(&self.user_id);
141    }
142}
143
144impl std::fmt::Debug for BillingService {
145    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
146        f.debug_struct("BillingService")
147            .field("provider", &self.provider.name())
148            .finish_non_exhaustive()
149    }
150}
151
152impl BillingService {
153    /// Build a service. Tests use this to wire a fake provider and a memory store.
154    #[must_use]
155    pub fn new(
156        provider: Arc<dyn BillingProvider>,
157        store: Arc<dyn BillingStore>,
158        catalog: PlanCatalog,
159        config: BillingConfig,
160        hooks: Arc<dyn BillingHooks>,
161    ) -> Self {
162        Self {
163            provider,
164            store,
165            catalog: Arc::new(catalog),
166            config: Arc::new(config),
167            hooks,
168            checkouts_in_flight: std::sync::Mutex::default(),
169        }
170    }
171
172    /// Claim the checkout slot for `user_id`. It returns `None` when one is in progress.
173    ///
174    /// The slot stays held from the live-subscription check to the end of the provider call.
175    /// It is per process. Replicas do not share it.
176    pub(crate) fn begin_checkout(self: &Arc<Self>, user_id: &str) -> Option<CheckoutSlot> {
177        let claimed = self
178            .checkouts_in_flight
179            .lock()
180            .unwrap_or_else(std::sync::PoisonError::into_inner)
181            .insert(user_id.to_owned());
182        claimed.then(|| CheckoutSlot {
183            service: Arc::clone(self),
184            user_id: user_id.to_owned(),
185        })
186    }
187
188    /// The service installed on `state`, if the plugin started.
189    #[must_use]
190    pub fn from_state(state: &AppState) -> Option<Arc<Self>> {
191        state.extension::<Self>()
192    }
193
194    /// The service, or a 503 error when the plugin did not start.
195    ///
196    /// # Errors
197    ///
198    /// Returns `AutumnError::service_unavailable` when the plugin did not start.
199    pub fn require(state: &AppState) -> Result<Arc<Self>, AutumnError> {
200        Self::from_state(state)
201            .ok_or_else(|| AutumnError::service_unavailable_msg("billing plugin is not started"))
202    }
203
204    /// The provider.
205    #[must_use]
206    pub fn provider(&self) -> &Arc<dyn BillingProvider> {
207        &self.provider
208    }
209
210    /// The mirror store.
211    #[must_use]
212    pub fn store(&self) -> &Arc<dyn BillingStore> {
213        &self.store
214    }
215
216    /// The plan catalog.
217    #[must_use]
218    pub fn catalog(&self) -> &PlanCatalog {
219        &self.catalog
220    }
221
222    /// The config.
223    #[must_use]
224    pub fn config(&self) -> &BillingConfig {
225        &self.config
226    }
227
228    /// The hooks.
229    #[must_use]
230    pub fn hooks(&self) -> &Arc<dyn BillingHooks> {
231        &self.hooks
232    }
233}
234
235/// The billing plugin. Mount with `app.plugin(BillingPlugin::new()...)`.
236pub struct BillingPlugin {
237    config: BillingConfig,
238    catalog: PlanCatalog,
239    provider: Option<Arc<dyn BillingProvider>>,
240    store: Option<Arc<dyn BillingStore>>,
241    hooks: Arc<dyn BillingHooks>,
242}
243
244impl std::fmt::Debug for BillingPlugin {
245    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
246        f.debug_struct("BillingPlugin")
247            .field("config", &self.config)
248            .field("plans", &self.catalog)
249            .finish_non_exhaustive()
250    }
251}
252
253impl Default for BillingPlugin {
254    fn default() -> Self {
255        Self::new()
256    }
257}
258
259impl BillingPlugin {
260    /// Defaults: Stripe from the environment, plans from config, store from the pool.
261    #[must_use]
262    pub fn new() -> Self {
263        Self {
264            config: BillingConfig::from_env(),
265            catalog: PlanCatalog::new(),
266            provider: None,
267            store: None,
268            hooks: Arc::new(NoHooks),
269        }
270    }
271
272    /// Supply the resolved `[billing]` config.
273    #[must_use]
274    pub fn config(mut self, config: BillingConfig) -> Self {
275        self.config = config;
276        self
277    }
278
279    /// Supply plans declared in code. Merged with `[[billing.plans]]`.
280    #[must_use]
281    pub fn plans(mut self, catalog: &PlanCatalog) -> Self {
282        for plan in catalog.plans() {
283            self.catalog.insert(plan.clone());
284        }
285        self
286    }
287
288    /// Replace the provider (default: Stripe).
289    #[must_use]
290    pub fn provider(mut self, provider: Arc<dyn BillingProvider>) -> Self {
291        self.provider = Some(provider);
292        self
293    }
294
295    /// Replace the store (default: the database when a pool exists, else memory).
296    #[must_use]
297    pub fn store(mut self, store: Arc<dyn BillingStore>) -> Self {
298        self.store = Some(store);
299        self
300    }
301
302    /// Install application hooks.
303    #[must_use]
304    pub fn hooks(mut self, hooks: Arc<dyn BillingHooks>) -> Self {
305        self.hooks = hooks;
306        self
307    }
308
309    fn resolve_store(
310        state: &AppState,
311        config: &BillingConfig,
312        injected: Option<Arc<dyn BillingStore>>,
313    ) -> Result<Arc<dyn BillingStore>, BillingError> {
314        if let Some(store) = injected {
315            return Ok(store);
316        }
317        #[cfg(feature = "db")]
318        if let Some(pool) = autumn_web::db::DbState::pool(state).cloned() {
319            return Ok(Arc::new(DbBillingStore::new(pool)));
320        }
321        let is_production = is_production_profile(state.profile());
322        check_memory_store(is_production, config)?;
323        if state.profile() != "test" {
324            tracing::warn!(
325                "🍂 Autumn Billing: no database pool; using the in-memory mirror, which is lost on restart"
326            );
327        }
328        Ok(Arc::new(MemoryBillingStore::new()))
329    }
330
331    fn resolve_provider(
332        state: &AppState,
333        config: &BillingConfig,
334        injected: Option<Arc<dyn BillingProvider>>,
335    ) -> Result<Arc<dyn BillingProvider>, BillingError> {
336        if let Some(provider) = injected {
337            return Ok(provider);
338        }
339        Ok(Arc::new(StripeProvider::from_state(state, &config.stripe)?))
340    }
341}
342
343impl Plugin for BillingPlugin {
344    fn name(&self) -> Cow<'static, str> {
345        Cow::Borrowed("autumn-billing")
346    }
347
348    fn contract(&self) -> Option<PluginContract> {
349        Some(autumn_web::plugin_contract::lockstep_contract(
350            env!("CARGO_PKG_NAME"),
351            env!("CARGO_PKG_VERSION"),
352        ))
353    }
354
355    fn build(self, app: AppBuilder) -> AppBuilder {
356        let Self {
357            mut config,
358            mut catalog,
359            provider,
360            store,
361            hooks,
362        } = self;
363        for plan in config.plans.drain(..) {
364            catalog.insert(plan);
365        }
366
367        let app = app.config_section("billing");
368        let app = app
369            .nest(&config.route_prefix, routes::router())
370            .declare_plugin_routes(routes::route_infos(&config));
371        let app = app.jobs(dunning::job_infos());
372        #[cfg(feature = "db")]
373        let app = app.plugin_migrations("autumn-billing", MIGRATIONS);
374
375        app.on_startup(move |state| {
376            let config = config.clone();
377            let catalog = catalog.clone();
378            let provider = provider.clone();
379            let store = store.clone();
380            let hooks = hooks.clone();
381            async move {
382                let is_production = is_production_profile(state.profile());
383                config.validate(is_production)?;
384                let provider = Self::resolve_provider(&state, &config, provider)?;
385                let store = Self::resolve_store(&state, &config, store)?;
386                verify_webhook_endpoint(&state, &config, provider.as_ref())?;
387                // `insert_extension` wraps the value in its own `Arc`, keyed
388                // by `BillingService`; read that handle back for the re-arm.
389                state.insert_extension(BillingService {
390                    provider,
391                    store,
392                    catalog: Arc::new(catalog),
393                    config: Arc::new(config),
394                    hooks,
395                    checkouts_in_flight: std::sync::Mutex::default(),
396                });
397                let service = BillingService::require(&state)?;
398                dunning::rearm_pending(state.clone(), service);
399                Ok(())
400            }
401        })
402    }
403}
404
405/// `true` for the production profile names.
406fn is_production_profile(profile: &str) -> bool {
407    matches!(profile, "prod" | "production")
408}
409
410/// The in-memory mirror is lost on restart. Production refuses it unless
411/// the config opts in.
412fn check_memory_store(is_production: bool, config: &BillingConfig) -> Result<(), BillingError> {
413    if is_production && !config.allow_memory_store_in_production {
414        return Err(BillingError::Config(
415            "autumn-billing: no database pool in production; the in-memory mirror is lost \
416             on restart. Configure a database, or set \
417             billing.allow_memory_store_in_production = true to accept this."
418                .to_owned(),
419        ));
420    }
421    Ok(())
422}
423
424/// Fail boot when the app did not declare the webhook receiver, or declared
425/// it with another verification preset than `provider` needs.
426fn verify_webhook_endpoint(
427    state: &AppState,
428    config: &BillingConfig,
429    provider: &dyn BillingProvider,
430) -> Result<(), AutumnError> {
431    let path = config.webhook_path();
432    let expected = provider
433        .webhook_endpoint(&config.endpoint_name, &path)
434        .map_err(BillingError::into_autumn)?;
435    let preset = expected.provider.as_str();
436    let config_arc = state.config_arc();
437    let declared = config_arc
438        .security
439        .webhooks
440        .endpoints
441        .iter()
442        .find(|endpoint| endpoint.path == path);
443    match declared {
444        Some(endpoint)
445            if endpoint.provider == expected.provider
446                && endpoint.max_body_bytes < expected.max_body_bytes =>
447        {
448            Err(AutumnError::internal_server_error_msg(format!(
449                "autumn-billing: the signed webhook endpoint at {path} allows request bodies of \
450                 {} bytes, but billing events can be larger (invoices with many lines) and an \
451                 oversized body is rejected before it is reconciled, so the provider retries it \
452                 forever. Set max_body_bytes = {} on the [[security.webhooks.endpoints]] entry \
453                 in autumn.toml.",
454                endpoint.max_body_bytes, expected.max_body_bytes
455            )))
456        }
457        Some(endpoint) if endpoint.provider == expected.provider => Ok(()),
458        Some(endpoint) => Err(AutumnError::internal_server_error_msg(format!(
459            "autumn-billing: the signed webhook endpoint at {path} declares provider = \"{}\", \
460             but the {} billing provider needs provider = \"{preset}\". Fix the \
461             [[security.webhooks.endpoints]] entry in autumn.toml.",
462            endpoint.provider.as_str(),
463            provider.name()
464        ))),
465        None => Err(AutumnError::internal_server_error_msg(format!(
466            "autumn-billing: no signed webhook endpoint is declared at {path}. Add to autumn.toml:\n\
467             [[security.webhooks.endpoints]]\n\
468             name = \"{}\"\n\
469             path = \"{path}\"\n\
470             provider = \"{preset}\"\n\
471             secret_env = \"{}_WEBHOOK_SECRET\"\n\
472             max_body_bytes = {}",
473            config.endpoint_name,
474            provider.name().to_uppercase(),
475            expected.max_body_bytes
476        ))),
477    }
478}
479
480// Route-conformance harness: the same checks `autumn-media-plugin`'s
481// `conformance_tests` uses, so the plugin's naming, prefix and attribution
482// conventions stay clean.
483#[cfg(test)]
484mod conformance_tests {
485    use autumn_web::plugin::Plugin;
486    use autumn_web::plugin_conformance::{
487        CheckStatus, ConformanceConfig, check_collisions, check_duplicate_registration,
488        check_route_attribution, check_route_prefix, check_sensitive_surfaces, run_conformance,
489    };
490    use autumn_web::route_listing::{RouteInfo, RouteSource};
491
492    use super::{BillingConfig, BillingPlugin, routes};
493
494    const PLUGIN_NAME: &str = "autumn-billing";
495    const PREFIX: &str = "/billing";
496
497    /// The routes `build` declares, attributed exactly as
498    /// `declare_plugin_routes` attributes them at runtime.
499    fn declared_routes() -> Vec<RouteInfo> {
500        routes::route_infos(&BillingConfig::default())
501    }
502
503    #[test]
504    fn build_declares_the_four_routes_under_the_prefix() {
505        let routes = declared_routes();
506        assert_eq!(routes.len(), 4, "the plugin declares exactly four routes");
507        assert!(
508            routes.iter().all(|route| route.path.starts_with(PREFIX)),
509            "every declared route lives under the prefix"
510        );
511        assert!(
512            routes
513                .iter()
514                .all(|route| route.source == RouteSource::Plugin(PLUGIN_NAME.to_owned())),
515            "every declared route is attributed to the plugin"
516        );
517    }
518
519    #[test]
520    fn routes_are_attributed_to_plugin_name() {
521        let result = check_route_attribution(PLUGIN_NAME, &declared_routes());
522        assert_eq!(
523            result.status,
524            CheckStatus::Pass,
525            "attribution failed: {}",
526            result.message
527        );
528    }
529
530    #[test]
531    fn routes_live_under_prefix() {
532        let result = check_route_prefix(PLUGIN_NAME, PREFIX, &[], &declared_routes());
533        assert_eq!(
534            result.status,
535            CheckStatus::Pass,
536            "prefix check failed: {}",
537            result.message
538        );
539    }
540
541    #[test]
542    fn routes_have_no_collisions_in_isolation() {
543        let (result, _) = check_collisions(&declared_routes());
544        assert_eq!(
545            result.status,
546            CheckStatus::Pass,
547            "unexpected collision: {}",
548            result.message
549        );
550    }
551
552    #[test]
553    fn routes_have_no_undeclared_sensitive_surfaces() {
554        let result = check_sensitive_surfaces(PLUGIN_NAME, &declared_routes(), &[]);
555        assert_eq!(
556            result.status,
557            CheckStatus::Pass,
558            "sensitive-surfaces check failed: {}",
559            result.message
560        );
561    }
562
563    #[test]
564    fn single_registration_passes_duplicate_check() {
565        let result = check_duplicate_registration(PLUGIN_NAME, &declared_routes());
566        assert_eq!(
567            result.status,
568            CheckStatus::Pass,
569            "single registration should pass: {}",
570            result.message
571        );
572    }
573
574    #[test]
575    fn routes_pass_full_conformance() {
576        let contract = BillingPlugin::new().contract().expect("a contract");
577        let config = ConformanceConfig::new(PLUGIN_NAME)
578            .contract(contract)
579            .prefix(PREFIX);
580        let report = run_conformance(&config, &declared_routes());
581        assert!(
582            report.passed(),
583            "BillingPlugin conformance failed:\n{}",
584            report.to_text_report()
585        );
586    }
587
588    #[test]
589    fn duplicate_registration_detected() {
590        // Installing the plugin twice would double its routes.
591        let mut routes = declared_routes();
592        routes.extend(declared_routes());
593        let result = check_duplicate_registration(PLUGIN_NAME, &routes);
594        assert_eq!(
595            result.status,
596            CheckStatus::Fail,
597            "expected duplicate-registration FAIL when installed twice"
598        );
599    }
600
601    #[test]
602    fn collision_with_host_route_detected() {
603        let mut routes = declared_routes();
604        routes.push(RouteInfo {
605            method: "POST".to_owned(),
606            path: format!("{PREFIX}/checkout"),
607            handler: "host::checkout".to_owned(),
608            source: RouteSource::User,
609            ..Default::default()
610        });
611        let (result, _) = check_collisions(&routes);
612        assert_eq!(
613            result.status,
614            CheckStatus::Fail,
615            "expected collision to be detected"
616        );
617    }
618}
619
620#[cfg(test)]
621mod store_resolution_tests {
622    use super::{BillingConfig, BillingError, check_memory_store, is_production_profile};
623
624    #[test]
625    fn production_refuses_the_memory_mirror_unless_allowed() {
626        let default = BillingConfig::default();
627        let err = check_memory_store(true, &default).expect_err("production refuses memory");
628        let BillingError::Config(message) = err else {
629            panic!("expected a config error");
630        };
631        assert!(
632            message.contains("allow_memory_store_in_production"),
633            "{message}"
634        );
635        check_memory_store(false, &default).expect("development accepts memory");
636        let allowed = BillingConfig::default().allow_memory_store_in_production(true);
637        check_memory_store(true, &allowed).expect("opt-in accepts memory");
638    }
639
640    #[test]
641    fn production_profile_names() {
642        assert!(is_production_profile("prod"));
643        assert!(is_production_profile("production"));
644        assert!(!is_production_profile("test"));
645        assert!(!is_production_profile("dev"));
646    }
647}
648
649#[cfg(test)]
650mod contract_tests {
651    use super::BillingPlugin;
652    use autumn_web::plugin::Plugin;
653
654    #[test]
655    fn contract_declares_lockstep_with_own_crate() {
656        let contract = BillingPlugin::new().contract().expect("a contract");
657        assert_eq!(contract.plugin, env!("CARGO_PKG_NAME"));
658        assert_eq!(
659            contract.autumn_web.as_deref(),
660            Some(autumn_web::plugin_contract::lockstep_range(env!("CARGO_PKG_VERSION")).as_str())
661        );
662        assert!(contract.experimental_surfaces.is_empty());
663    }
664}
autumn_billing/lib.rs

autumn_billing/
lib.rs
