//! Pluggable subscription storage. //! //! Same shape as the other domain stores: a sync trait plus an //! [`InMemorySubscriptionStore`] reference impl. A graph-backed //! impl lives in `op-graph`. use std::collections::HashMap; use std::sync::Mutex; use crate::error::{Error, Result}; use crate::subscription::{Subscription, SubscriptionId}; /// Pluggable subscription storage interface. pub trait SubscriptionStore: Send + Sync { /// Persist a fresh subscription. Idempotent on `external_id`. /// /// # Errors /// See [`Error`]. fn create_subscription(&self, s: Subscription) -> Result; /// Fetch by id. /// /// # Errors /// [`Error::NotFound`] if unknown. fn get_subscription(&self, id: SubscriptionId) -> Result; /// Find by external id. /// /// # Errors /// Backend-specific. fn find_by_external_id(&self, external_id: &str) -> Result>; /// All subscriptions for a customer. Order unspecified. /// /// # Errors /// Backend-specific. fn list_for_customer(&self, customer_ref: &str) -> Result>; /// All non-terminal subscriptions whose /// `current_period_end_unix_secs <= as_of`. Used by the /// billing tick to fan out due-charge events. /// /// # Errors /// Backend-specific. fn list_due_at(&self, as_of_unix_secs: u64) -> Result>; /// Apply a state transition via closure. Commits only on /// `Ok(())`. /// /// # Errors /// `Error::NotFound`, plus whatever the closure returns. fn update(&self, id: SubscriptionId, f: F) -> Result where F: FnOnce(&mut Subscription) -> Result<()>; } // ============================================================ // In-memory ref impl // ============================================================ #[derive(Default)] struct Inner { subs: HashMap, by_external_id: HashMap, } /// In-process subscription store. Not for multi-instance production. #[derive(Default)] pub struct InMemorySubscriptionStore { inner: Mutex, } impl InMemorySubscriptionStore { /// Construct empty. #[must_use] pub fn new() -> Self { Self::default() } /// Number of subscriptions tracked. /// /// # Panics /// Panics only if the internal mutex was previously poisoned /// by a thread that panicked while holding it. #[must_use] pub fn len(&self) -> usize { self.inner.lock().expect("poisoned").subs.len() } /// True if no subscriptions stored. #[must_use] pub fn is_empty(&self) -> bool { self.len() == 0 } } fn bodies_equivalent(a: &Subscription, b: &Subscription) -> bool { a.customer_ref == b.customer_ref && a.plan.id == b.plan.id && a.plan.amount == b.plan.amount && a.plan.interval == b.plan.interval && a.plan.interval_count == b.plan.interval_count && a.external_id == b.external_id } impl SubscriptionStore for InMemorySubscriptionStore { fn create_subscription(&self, s: Subscription) -> Result { let mut inner = self.inner.lock().expect("poisoned"); if let Some(ext) = &s.external_id && let Some(existing_id) = inner.by_external_id.get(ext).copied() { let existing = inner .subs .get(&existing_id) .expect("index/store invariant") .clone(); if bodies_equivalent(&existing, &s) { return Ok(existing_id); } return Err(Error::IdempotencyMismatch(ext.clone())); } if let Some(ext) = &s.external_id { inner.by_external_id.insert(ext.clone(), s.id); } let id = s.id; inner.subs.insert(id, s); Ok(id) } fn get_subscription(&self, id: SubscriptionId) -> Result { self.inner .lock() .expect("poisoned") .subs .get(&id) .cloned() .ok_or_else(|| Error::NotFound(id.to_string())) } fn find_by_external_id(&self, external_id: &str) -> Result> { let inner = self.inner.lock().expect("poisoned"); let Some(id) = inner.by_external_id.get(external_id).copied() else { return Ok(None); }; Ok(inner.subs.get(&id).cloned()) } fn list_for_customer(&self, customer_ref: &str) -> Result> { Ok(self .inner .lock() .expect("poisoned") .subs .values() .filter(|s| s.customer_ref == customer_ref) .cloned() .collect()) } fn list_due_at(&self, as_of_unix_secs: u64) -> Result> { Ok(self .inner .lock() .expect("poisoned") .subs .values() .filter(|s| { !s.status.is_terminal() && s.current_period_end_unix_secs <= as_of_unix_secs }) .cloned() .collect()) } fn update(&self, id: SubscriptionId, f: F) -> Result where F: FnOnce(&mut Subscription) -> Result<()>, { let mut inner = self.inner.lock().expect("poisoned"); let existing = inner .subs .get(&id) .ok_or_else(|| Error::NotFound(id.to_string()))?; let mut staged = existing.clone(); f(&mut staged)?; let result = staged.clone(); inner.subs.insert(id, staged); Ok(result) } } #[cfg(test)] mod tests { use super::*; use crate::plan::{Interval, Plan}; use op_core::{Currency, Money, PaymentMethod, VaultRef}; fn plan() -> Plan { Plan::new( "p", Money::from_minor(1000, Currency::USD), Interval::Month, 1, ) .unwrap() } fn sub(customer: &str, ext: Option<&str>) -> Subscription { let mut s = Subscription::new( customer, plan(), PaymentMethod::Vault(VaultRef::new("tok")), 1_700_000_000, ) .unwrap(); if let Some(e) = ext { s = s.with_external_id(e); } s } #[test] fn round_trip() { let store = InMemorySubscriptionStore::new(); let s = sub("c-1", None); let id = s.id; store.create_subscription(s.clone()).unwrap(); let got = store.get_subscription(id).unwrap(); assert_eq!(got.id, id); assert_eq!(got.customer_ref, s.customer_ref); assert_eq!(got.plan.id, s.plan.id); assert_eq!(got.status.code(), s.status.code()); } #[test] fn idempotency_same_body() { let store = InMemorySubscriptionStore::new(); let s = sub("c-1", Some("e1")); let id1 = store.create_subscription(s.clone()).unwrap(); let id2 = store.create_subscription(s).unwrap(); assert_eq!(id1, id2); } #[test] fn idempotency_mismatch_rejects() { let store = InMemorySubscriptionStore::new(); store.create_subscription(sub("c-1", Some("e1"))).unwrap(); assert!(matches!( store .create_subscription(sub("c-2", Some("e1"))) .unwrap_err(), Error::IdempotencyMismatch(_) )); } #[test] fn list_for_customer_filters() { let store = InMemorySubscriptionStore::new(); store.create_subscription(sub("c-1", Some("a"))).unwrap(); store.create_subscription(sub("c-1", Some("b"))).unwrap(); store.create_subscription(sub("c-2", Some("c"))).unwrap(); assert_eq!(store.list_for_customer("c-1").unwrap().len(), 2); assert_eq!(store.list_for_customer("c-2").unwrap().len(), 1); assert_eq!(store.list_for_customer("nope").unwrap().len(), 0); } #[test] fn list_due_at_filters_by_period_end() { let store = InMemorySubscriptionStore::new(); let s = sub("c-1", None); let period_end = s.current_period_end_unix_secs; store.create_subscription(s).unwrap(); assert!(store.list_due_at(period_end - 1).unwrap().is_empty()); assert_eq!(store.list_due_at(period_end + 1).unwrap().len(), 1); } #[test] fn update_commits_on_ok_only() { let store = InMemorySubscriptionStore::new(); let s = sub("c-1", None); let id = s.id; store.create_subscription(s).unwrap(); store.update(id, |s| s.pause(2_000)).unwrap(); assert_eq!(store.get_subscription(id).unwrap().status.code(), "paused"); let _ = store .update(id, |s| { s.resume()?; Err(Error::Invalid("rollback".into())) }) .unwrap_err(); // Closure failed; resume() side effect was discarded. assert_eq!(store.get_subscription(id).unwrap().status.code(), "paused"); } }