rovenue/virtual_currencies/
reader.rs

1use std::collections::BTreeMap;
2use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
3use std::sync::Arc;
4
5use crate::cache::{CacheStore, VirtualCurrencyRepo};
6use crate::error::{RovenueError, RovenueResult};
7use crate::identity::IdentityManager;
8use crate::observer::{ChangeEvent, ObserverBus};
9use crate::time::Clock;
10use crate::transport::api::ApiEnvelope;
11use crate::transport::http_client::HttpClient;
12use crate::transport::types::HttpRequest;
13
14use super::types::VcBalancesWire;
15
16pub struct VirtualCurrencyReader {
17    store: Arc<CacheStore>,
18    identity: Arc<IdentityManager>,
19    http: Option<Arc<HttpClient>>,
20    bus: Option<Arc<ObserverBus>>,
21    clock: Option<Arc<dyn Clock>>,
22    last_refresh_ms: AtomicU64,
23    refreshing: AtomicBool,
24}
25
26impl VirtualCurrencyReader {
27    pub fn new(store: Arc<CacheStore>, identity: Arc<IdentityManager>) -> Self {
28        Self {
29            store,
30            identity,
31            http: None,
32            bus: None,
33            clock: None,
34            last_refresh_ms: AtomicU64::new(0),
35            refreshing: AtomicBool::new(false),
36        }
37    }
38    pub fn with_http(mut self, http: Arc<HttpClient>) -> Self {
39        self.http = Some(http);
40        self
41    }
42    pub fn with_observer_bus(mut self, bus: Arc<ObserverBus>) -> Self {
43        self.bus = Some(bus);
44        self
45    }
46    pub fn with_clock(mut self, clock: Arc<dyn Clock>) -> Self {
47        self.clock = Some(clock);
48        self
49    }
50
51    fn repo(&self) -> VirtualCurrencyRepo {
52        VirtualCurrencyRepo::new(Arc::clone(&self.store))
53    }
54
55    /// All cached balances for the current user scope (code → balance).
56    pub fn balances(&self) -> BTreeMap<String, i64> {
57        let scope = self.identity.current_user_scope();
58        self.repo().get_all(&scope).unwrap_or_default()
59    }
60
61    /// One currency's cached balance, or 0 when absent.
62    pub fn balance(&self, code: &str) -> i64 {
63        let scope = self.identity.current_user_scope();
64        self.repo().get(&scope, code).ok().flatten().unwrap_or(0)
65    }
66
67    /// GET /v1/virtual-currencies/me → replace the cached balance set.
68    pub fn refresh(&self) -> RovenueResult<()> {
69        let http = self.http.as_ref().ok_or(RovenueError::Internal())?;
70        let clock = self.clock.as_ref().ok_or(RovenueError::Internal())?;
71        // Cache namespaced by scope; wire identity is always the stable
72        // rovenue_id (server resolves the header as a rovenueId).
73        let scope = self.identity.current_user_scope();
74        let wire_id = self.identity.rovenue_id();
75
76        let resp = http.get_json::<ApiEnvelope<VcBalancesWire>>(
77            HttpRequest::new("/v1/virtual-currencies/me").user_scope(&wire_id),
78        )?;
79        let body = resp.body.ok_or(RovenueError::Internal())?;
80        let balances: BTreeMap<String, i64> = body.data.balances.into_iter().collect();
81        self.set_balances(&scope, &balances, clock.now_unix_ms())
82    }
83
84    /// Persist a balance set and emit `VirtualCurrenciesChanged` if it changed.
85    /// Stamps freshness so background coalescing (`maybe_refresh_async`) settles.
86    pub fn set_balances(
87        &self,
88        scope: &str,
89        balances: &BTreeMap<String, i64>,
90        now_ms: u64,
91    ) -> RovenueResult<()> {
92        let repo = self.repo();
93        let changed = repo
94            .get_all(scope)
95            .map(|prev| &prev != balances)
96            .unwrap_or(true);
97        repo.upsert_all(scope, balances, now_ms)?;
98        if changed {
99            if let Some(bus) = &self.bus {
100                bus.emit(ChangeEvent::VirtualCurrenciesChanged);
101            }
102        }
103        self.last_refresh_ms.store(now_ms, Ordering::Relaxed);
104        Ok(())
105    }
106
107    fn is_stale(&self, now: u64, staleness_ms: u64) -> bool {
108        now.saturating_sub(self.last_refresh_ms.load(Ordering::Relaxed)) > staleness_ms
109    }
110
111    pub fn maybe_refresh_async(self: &std::sync::Arc<Self>, staleness_ms: u64) {
112        let now = match &self.clock {
113            Some(c) => c.now_unix_ms(),
114            None => return,
115        };
116        if !self.is_stale(now, staleness_ms) {
117            return;
118        }
119        if self
120            .refreshing
121            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
122            .is_err()
123        {
124            return;
125        }
126        let this = std::sync::Arc::clone(self);
127        std::thread::spawn(move || {
128            // Clear the coalesce flag even if refresh() unwinds, so a panicking
129            // observer callback can't permanently disable background refresh.
130            struct ClearOnDrop<'a>(&'a std::sync::atomic::AtomicBool);
131            impl Drop for ClearOnDrop<'_> {
132                fn drop(&mut self) {
133                    self.0.store(false, std::sync::atomic::Ordering::Release);
134                }
135            }
136            let _guard = ClearOnDrop(&this.refreshing);
137            let _ = this.refresh();
138        });
139    }
140}
rovenue/virtual_currencies/reader.rs

rovenue/virtual_currencies/
reader.rs
