#[cfg(not(target_arch = "wasm32"))]
use std::sync::Mutex;
use std::{
collections::{BTreeMap, BTreeSet},
sync::{
Arc, OnceLock,
atomic::{AtomicU64, Ordering},
},
};
use crate::registry::{RecoveryGate, ServiceRegistry};
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Product {
pub id: String,
pub display_price: String,
pub title: String,
pub description: String,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum StorePhase {
#[default]
Unavailable,
Blocked,
Connecting,
Ready,
}
impl StorePhase {
pub fn cannot_sell(self) -> bool {
matches!(self, Self::Unavailable | Self::Blocked)
}
pub fn may_yet_change(self) -> bool {
matches!(self, Self::Unavailable | Self::Connecting)
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct StoreState {
pub phase: StorePhase,
pub products: Vec<Product>,
pub owned: BTreeSet<String>,
pub orders: BTreeMap<String, String>,
pub error: Option<String>,
pub busy: bool,
}
impl StoreState {
pub fn owns(&self, product_id: &str) -> bool {
self.owned.contains(product_id)
}
pub fn order_id(&self, product_id: &str) -> Option<&str> {
self.orders.get(product_id).map(String::as_str)
}
pub fn product(&self, product_id: &str) -> Option<&Product> {
self.products.iter().find(|p| p.id == product_id)
}
pub fn display_price(&self, product_id: &str) -> Option<&str> {
self.product(product_id).map(|p| p.display_price.as_str())
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum PurchaseEvent {
Purchased(String),
Cancelled,
Pending,
Failed(String),
Restored {
restored: usize,
},
}
pub trait Purchases: Send + Sync {
fn configure(&self, product_ids: &[&str]);
fn state(&self) -> StoreState;
fn purchase(&self, product_id: &str);
fn restore(&self);
fn take_event(&self) -> Option<PurchaseEvent>;
fn is_connected(&self) -> bool;
fn reconnect(&self);
}
pub type PurchasesRef = Arc<dyn Purchases>;
struct NoPurchases;
impl Purchases for NoPurchases {
fn configure(&self, _product_ids: &[&str]) {}
fn state(&self) -> StoreState {
StoreState::default()
}
fn purchase(&self, _product_id: &str) {}
fn restore(&self) {}
fn take_event(&self) -> Option<PurchaseEvent> {
None
}
fn is_connected(&self) -> bool {
false
}
fn reconnect(&self) {}
}
static PLATFORM_PURCHASES: ServiceRegistry<dyn Purchases> = ServiceRegistry::new();
static NO_PURCHASES: OnceLock<PurchasesRef> = OnceLock::new();
static DEFAULT_PURCHASES: OnceLock<PurchasesRef> = OnceLock::new();
static PURCHASE_RECOVERY: RecoveryGate = RecoveryGate::new();
struct PlatformPurchases;
fn registered_purchases() -> PurchasesRef {
PLATFORM_PURCHASES
.get_or_warn("purchases")
.unwrap_or_else(|| NO_PURCHASES.get_or_init(|| Arc::new(NoPurchases)).clone())
}
fn active_purchases() -> PurchasesRef {
let purchases = registered_purchases();
if purchases.is_connected() {
PURCHASE_RECOVERY.succeeded();
} else if PURCHASE_RECOVERY.try_start() {
purchases.reconnect();
}
purchases
}
impl Purchases for PlatformPurchases {
fn configure(&self, product_ids: &[&str]) {
active_purchases().configure(product_ids);
}
fn state(&self) -> StoreState {
active_purchases().state()
}
fn purchase(&self, product_id: &str) {
active_purchases().purchase(product_id);
}
fn restore(&self) {
active_purchases().restore();
}
fn take_event(&self) -> Option<PurchaseEvent> {
active_purchases().take_event()
}
fn is_connected(&self) -> bool {
registered_purchases().is_connected()
}
fn reconnect(&self) {
registered_purchases().reconnect();
}
}
#[cfg(not(target_arch = "wasm32"))]
type StoreListener = Arc<dyn Fn() + Send + Sync>;
#[cfg(target_arch = "wasm32")]
type StoreListener = std::rc::Rc<dyn Fn()>;
#[cfg(not(target_arch = "wasm32"))]
fn store_listeners() -> &'static Mutex<Vec<(u64, StoreListener)>> {
static LISTENERS: OnceLock<Mutex<Vec<(u64, StoreListener)>>> = OnceLock::new();
LISTENERS.get_or_init(|| Mutex::new(Vec::new()))
}
#[cfg(target_arch = "wasm32")]
thread_local! {
static STORE_LISTENERS: std::cell::RefCell<Vec<(u64, StoreListener)>> = const { std::cell::RefCell::new(Vec::new()) };
}
static NEXT_STORE_LISTENER_ID: AtomicU64 = AtomicU64::new(1);
pub struct StoreObserver {
id: u64,
}
impl Drop for StoreObserver {
fn drop(&mut self) {
#[cfg(not(target_arch = "wasm32"))]
if let Ok(mut listeners) = store_listeners().lock() {
listeners.retain(|(id, _)| *id != self.id);
}
#[cfg(target_arch = "wasm32")]
STORE_LISTENERS.with(|listeners| listeners.borrow_mut().retain(|(id, _)| *id != self.id));
}
}
#[cfg(not(target_arch = "wasm32"))]
pub fn observe_store_news(listener: impl Fn() + Send + Sync + 'static) -> StoreObserver {
let id = NEXT_STORE_LISTENER_ID.fetch_add(1, Ordering::Relaxed);
if let Ok(mut listeners) = store_listeners().lock() {
listeners.push((id, Arc::new(listener)));
}
StoreObserver { id }
}
#[cfg(target_arch = "wasm32")]
pub fn observe_store_news(listener: impl Fn() + 'static) -> StoreObserver {
let id = NEXT_STORE_LISTENER_ID.fetch_add(1, Ordering::Relaxed);
STORE_LISTENERS.with(|listeners| {
listeners
.borrow_mut()
.push((id, std::rc::Rc::new(listener)));
});
StoreObserver { id }
}
pub fn note_store_news() {
#[cfg(not(target_arch = "wasm32"))]
let listeners = store_listeners()
.lock()
.map(|listeners| {
listeners
.iter()
.map(|(_, listener)| Arc::clone(listener))
.collect::<Vec<_>>()
})
.unwrap_or_default();
#[cfg(target_arch = "wasm32")]
let listeners = STORE_LISTENERS.with(|listeners| {
listeners
.borrow()
.iter()
.map(|(_, listener)| std::rc::Rc::clone(listener))
.collect::<Vec<_>>()
});
for listener in listeners {
listener();
}
}
pub fn set_platform_purchases(purchases: PurchasesRef) {
PLATFORM_PURCHASES.set(purchases);
PURCHASE_RECOVERY.succeeded();
}
pub fn clear_platform_purchases() {
PLATFORM_PURCHASES.clear();
}
pub fn purchases() -> PurchasesRef {
DEFAULT_PURCHASES
.get_or_init(|| Arc::new(PlatformPurchases))
.clone()
}
pub fn store_available() -> bool {
PLATFORM_PURCHASES.get().is_some()
}
pub fn configure(product_ids: &[&str]) {
purchases().configure(product_ids);
}
pub fn store_state() -> StoreState {
purchases().state()
}
pub fn purchase(product_id: &str) {
purchases().purchase(product_id);
}
pub fn restore() {
purchases().restore();
}
pub fn take_event() -> Option<PurchaseEvent> {
purchases().take_event()
}
#[allow(non_snake_case)]
#[track_caller]
pub fn rememberStoreState() -> cranpose_core::State<StoreState> {
let updates = cranpose_core::rememberEventStream((), |sender| {
observe_store_news(move || sender.send(store_state()))
});
cranpose_core::collectAsState(updates, (), store_state())
}
#[allow(non_snake_case)]
#[track_caller]
pub fn rememberPurchaseEvents() -> cranpose_core::EventStream<PurchaseEvent> {
cranpose_core::rememberEventStream((), |sender| {
observe_store_news(move || {
while let Some(event) = take_event() {
sender.send(event);
}
})
})
}
#[cfg(test)]
mod tests {
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use super::*;
#[test]
fn default_backend_sells_nothing_and_owns_nothing() {
let _guard = crate::registry::test_service_guard();
clear_platform_purchases();
let state = store_state();
assert_eq!(state.phase, StorePhase::Unavailable);
assert!(state.owned.is_empty());
assert!(!state.owns("com.example.pro"));
assert!(!store_available());
configure(&["com.example.pro"]);
purchase("com.example.pro");
restore();
assert_eq!(take_event(), None);
}
#[test]
fn the_two_phases_that_cannot_sell_differ_on_whether_waiting_helps() {
assert!(StorePhase::Unavailable.cannot_sell());
assert!(StorePhase::Blocked.cannot_sell());
assert!(!StorePhase::Connecting.cannot_sell());
assert!(!StorePhase::Ready.cannot_sell());
assert!(StorePhase::Unavailable.may_yet_change());
assert!(StorePhase::Connecting.may_yet_change());
assert!(
!StorePhase::Blocked.may_yet_change(),
"a store that has said no is what the phase exists to say"
);
assert!(!StorePhase::Ready.may_yet_change());
}
#[test]
fn nothing_is_owned_by_default_and_blocked_is_not_the_default() {
assert_eq!(StorePhase::default(), StorePhase::Unavailable);
}
#[test]
fn installed_backend_answers_prices_and_ownership() {
let _guard = crate::registry::test_service_guard();
struct Fake;
impl Purchases for Fake {
fn configure(&self, _product_ids: &[&str]) {}
fn state(&self) -> StoreState {
StoreState {
phase: StorePhase::Ready,
products: vec![Product {
id: "com.example.pro".into(),
display_price: "34,99 €".into(),
title: "Pro".into(),
description: "Everything unlocked".into(),
}],
owned: BTreeSet::from(["com.example.pro".to_string()]),
orders: BTreeMap::from([(
"com.example.pro".to_string(),
"GPA.1234-5678".to_string(),
)]),
error: None,
busy: false,
}
}
fn purchase(&self, _product_id: &str) {}
fn restore(&self) {}
fn take_event(&self) -> Option<PurchaseEvent> {
Some(PurchaseEvent::Purchased("com.example.pro".into()))
}
fn is_connected(&self) -> bool {
true
}
fn reconnect(&self) {}
}
set_platform_purchases(Arc::new(Fake));
let state = store_state();
assert_eq!(state.phase, StorePhase::Ready);
assert!(state.owns("com.example.pro"));
assert_eq!(state.order_id("com.example.pro"), Some("GPA.1234-5678"));
assert_eq!(state.order_id("com.example.free"), None);
assert_eq!(state.display_price("com.example.pro"), Some("34,99 €"));
assert_eq!(state.display_price("com.example.nope"), None);
assert!(store_available());
assert_eq!(
take_event(),
Some(PurchaseEvent::Purchased("com.example.pro".into()))
);
clear_platform_purchases();
}
#[test]
fn dead_store_reconnects_before_frame_state_is_read() {
let _guard = crate::registry::test_service_guard();
struct Reconnecting {
alive: AtomicBool,
reconnects: AtomicUsize,
}
impl Purchases for Reconnecting {
fn configure(&self, _product_ids: &[&str]) {}
fn state(&self) -> StoreState {
StoreState {
phase: if self.alive.load(Ordering::Acquire) {
StorePhase::Ready
} else {
StorePhase::Unavailable
},
..StoreState::default()
}
}
fn purchase(&self, _product_id: &str) {}
fn restore(&self) {}
fn take_event(&self) -> Option<PurchaseEvent> {
None
}
fn is_connected(&self) -> bool {
self.alive.load(Ordering::Acquire)
}
fn reconnect(&self) {
self.reconnects.fetch_add(1, Ordering::AcqRel);
self.alive.store(true, Ordering::Release);
}
}
clear_platform_purchases();
let purchases = Arc::new(Reconnecting {
alive: AtomicBool::new(false),
reconnects: AtomicUsize::new(0),
});
set_platform_purchases(purchases.clone());
assert_eq!(store_state().phase, StorePhase::Ready);
assert_eq!(purchases.reconnects.load(Ordering::Acquire), 1);
clear_platform_purchases();
}
#[test]
fn store_observers_receive_news_until_dropped() {
let calls = Arc::new(AtomicUsize::new(0));
let seen = Arc::clone(&calls);
let observer = observe_store_news(move || {
seen.fetch_add(1, Ordering::Relaxed);
});
note_store_news();
assert_eq!(calls.load(Ordering::Relaxed), 1);
drop(observer);
note_store_news();
assert_eq!(calls.load(Ordering::Relaxed), 1);
}
}