rovenue/remote_config/
reader.rs

1use std::collections::HashMap;
2use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
3use std::sync::{Arc, RwLock};
4
5use serde_json::Value;
6
7use crate::cache::remote_config::RemoteConfigCacheRepo;
8use crate::cache::CacheStore;
9use crate::error::{ErrorKind, RovenueError, RovenueResult};
10use crate::identity::IdentityManager;
11use crate::observer::{ChangeEvent, ObserverBus};
12use crate::time::{Clock, SystemClock};
13use crate::transport::api::ApiEnvelope;
14use crate::transport::http_client::HttpClient;
15use crate::transport::types::HttpRequest;
16
17use super::types::{ConfigResponse, ExperimentAssignment};
18
19/// Cache key for the (subscriber-scoped) Remote Config payload.
20const REMOTE_CONFIG_RESOURCE: &str = "remote_config";
21
22#[derive(Default)]
23struct State {
24    flags: serde_json::Map<String, Value>,
25    experiments: HashMap<String, ExperimentAssignment>,
26}
27
28struct Inner {
29    http: Arc<HttpClient>,
30    store: Arc<CacheStore>,
31    identity: Arc<IdentityManager>,
32    clock: Arc<dyn Clock>,
33    bus: Option<Arc<ObserverBus>>,
34    state: RwLock<State>,
35    last_refresh_ms: AtomicU64,
36    refreshing: AtomicBool,
37    hydrated: AtomicBool,
38}
39
40/// Reads Remote Config (feature flags + experiment assignments) from
41/// `GET /v1/config`, caching the last-known payload both in memory and on disk
42/// so flags resolve instantly and survive offline. Mirrors the offerings/
43/// entitlements readers: a 60s scheduler tick refreshes in the background and
44/// reads coalesce a single staleness-driven refresh.
45pub struct RemoteConfigReader {
46    inner: Arc<Inner>,
47}
48
49impl RemoteConfigReader {
50    pub fn new(
51        http: Arc<HttpClient>,
52        store: Arc<CacheStore>,
53        identity: Arc<IdentityManager>,
54    ) -> Self {
55        Self {
56            inner: Arc::new(Inner {
57                http,
58                store,
59                identity,
60                clock: Arc::new(SystemClock),
61                bus: None,
62                state: RwLock::new(State::default()),
63                last_refresh_ms: AtomicU64::new(0),
64                refreshing: AtomicBool::new(false),
65                hydrated: AtomicBool::new(false),
66            }),
67        }
68    }
69
70    pub fn with_clock(self, clock: Arc<dyn Clock>) -> Self {
71        // Inner is freshly constructed and not yet shared, so the Arc is unique.
72        let mut inner = Arc::try_unwrap(self.inner).unwrap_or_else(|arc| {
73            // Should never happen during construction; fall back to a clone of fields.
74            Inner {
75                http: Arc::clone(&arc.http),
76                store: Arc::clone(&arc.store),
77                identity: Arc::clone(&arc.identity),
78                clock: Arc::clone(&arc.clock),
79                bus: arc.bus.clone(),
80                state: RwLock::new(State::default()),
81                last_refresh_ms: AtomicU64::new(0),
82                refreshing: AtomicBool::new(false),
83                hydrated: AtomicBool::new(false),
84            }
85        });
86        inner.clock = clock;
87        Self {
88            inner: Arc::new(inner),
89        }
90    }
91
92    pub fn with_observer_bus(self, bus: Arc<ObserverBus>) -> Self {
93        let mut inner = Arc::try_unwrap(self.inner).unwrap_or_else(|arc| Inner {
94            http: Arc::clone(&arc.http),
95            store: Arc::clone(&arc.store),
96            identity: Arc::clone(&arc.identity),
97            clock: Arc::clone(&arc.clock),
98            bus: arc.bus.clone(),
99            state: RwLock::new(State::default()),
100            last_refresh_ms: AtomicU64::new(0),
101            refreshing: AtomicBool::new(false),
102            hydrated: AtomicBool::new(false),
103        });
104        inner.bus = Some(bus);
105        Self {
106            inner: Arc::new(inner),
107        }
108    }
109
110    /// Force a synchronous refresh from the server.
111    pub fn refresh(&self) -> RovenueResult<()> {
112        self.inner.refresh()
113    }
114
115    /// Trigger one coalesced background refresh if the cached payload is older
116    /// than `staleness_ms`. Never blocks the caller; multiple concurrent reads
117    /// collapse into a single in-flight GET.
118    pub fn maybe_refresh_async(&self, staleness_ms: u64) {
119        let now = self.inner.clock.now_unix_ms();
120        let last = self.inner.last_refresh_ms.load(Ordering::SeqCst);
121        if last != 0 && now.saturating_sub(last) < staleness_ms {
122            return;
123        }
124        // Claim the in-flight slot; if another refresh holds it, bail.
125        if self.inner.refreshing.swap(true, Ordering::SeqCst) {
126            return;
127        }
128        let inner = Arc::clone(&self.inner);
129        std::thread::spawn(move || {
130            let _ = inner.refresh();
131            inner.refreshing.store(false, Ordering::SeqCst);
132        });
133    }
134
135    fn flag_value(&self, key: &str) -> Option<Value> {
136        self.inner.ensure_hydrated();
137        let guard = self.inner.state.read().ok()?;
138        guard.flags.get(key).cloned()
139    }
140
141    pub fn bool(&self, key: &str, fallback: bool) -> bool {
142        match self.flag_value(key) {
143            Some(Value::Bool(b)) => b,
144            _ => fallback,
145        }
146    }
147
148    pub fn string(&self, key: &str, fallback: String) -> String {
149        match self.flag_value(key) {
150            Some(Value::String(s)) => s,
151            _ => fallback,
152        }
153    }
154
155    pub fn int(&self, key: &str, fallback: i64) -> i64 {
156        match self.flag_value(key) {
157            Some(Value::Number(n)) => n.as_i64().unwrap_or(fallback),
158            _ => fallback,
159        }
160    }
161
162    pub fn double(&self, key: &str, fallback: f64) -> f64 {
163        match self.flag_value(key) {
164            Some(Value::Number(n)) => n.as_f64().unwrap_or(fallback),
165            _ => fallback,
166        }
167    }
168
169    /// Raw JSON string for any present flag (primitive, object, or array).
170    /// `None` when the key is absent.
171    pub fn json(&self, key: &str) -> Option<String> {
172        self.flag_value(key).map(|v| v.to_string())
173    }
174
175    pub fn keys(&self) -> Vec<String> {
176        self.inner.ensure_hydrated();
177        let guard = match self.inner.state.read() {
178            Ok(g) => g,
179            Err(_) => return Vec::new(),
180        };
181        guard.flags.keys().cloned().collect()
182    }
183
184    pub fn experiment(&self, key: &str) -> Option<ExperimentAssignment> {
185        self.inner.ensure_hydrated();
186        let guard = self.inner.state.read().ok()?;
187        guard.experiments.get(key).cloned()
188    }
189
190    pub fn experiments_all(&self) -> Vec<ExperimentAssignment> {
191        self.inner.ensure_hydrated();
192        let guard = match self.inner.state.read() {
193            Ok(g) => g,
194            Err(_) => return Vec::new(),
195        };
196        guard.experiments.values().cloned().collect()
197    }
198
199    /// The entire cached payload as a JSON string:
200    /// `{ "flags": { key: value }, "experiments": { key: { experimentId, … } } }`.
201    /// Façades parse this once to back a reactive, synchronous hook surface
202    /// (e.g. RN `useRemoteConfig()`) without a bridge round-trip per read.
203    pub fn all_json(&self) -> String {
204        const EMPTY: &str = r#"{"flags":{},"experiments":{}}"#;
205        self.inner.ensure_hydrated();
206        let guard = match self.inner.state.read() {
207            Ok(g) => g,
208            Err(_) => return EMPTY.to_string(),
209        };
210        let experiments: serde_json::Map<String, Value> = guard
211            .experiments
212            .iter()
213            .map(|(k, e)| {
214                (
215                    k.clone(),
216                    serde_json::json!({
217                        "experimentId": e.experiment_id,
218                        "key": e.key,
219                        "variantId": e.variant_id,
220                        "variantName": e.variant_name,
221                        "value": serde_json::from_str::<Value>(&e.value_json)
222                            .unwrap_or(Value::Null),
223                    }),
224                )
225            })
226            .collect();
227        serde_json::json!({
228            "flags": Value::Object(guard.flags.clone()),
229            "experiments": Value::Object(experiments),
230        })
231        .to_string()
232    }
233}
234
235impl Inner {
236    /// Lazily load the persisted payload into memory exactly once, so a
237    /// just-launched (offline) app resolves flags from the last session.
238    fn ensure_hydrated(&self) {
239        if self.hydrated.swap(true, Ordering::SeqCst) {
240            return;
241        }
242        if let Ok(Some(raw)) = RemoteConfigCacheRepo::new(&self.store).get(REMOTE_CONFIG_RESOURCE) {
243            if let Ok(parsed) = serde_json::from_str::<ConfigResponse>(&raw) {
244                if let Ok(mut guard) = self.state.write() {
245                    *guard = build_state(parsed);
246                }
247            }
248        }
249    }
250
251    fn refresh(&self) -> RovenueResult<()> {
252        // Wire identity is always the stable rovenue_id — the server resolves
253        // the X-Rovenue-User-Id header as a rovenueId. (Config is cached under a
254        // fixed resource key, so there is no scope-namespaced cache to preserve.)
255        let wire_id = self.identity.rovenue_id();
256        let req = HttpRequest::new("/v1/config").subscriber_id(&wire_id);
257        match self.http.get_json::<ApiEnvelope<ConfigResponse>>(req) {
258            Ok(resp) => {
259                let body = resp.body.ok_or(RovenueError::Internal())?;
260                // Persist the raw payload first (best-effort) so an offline
261                // relaunch still has it even if the in-memory swap races.
262                if let Ok(raw) = serde_json::to_string(&body.data) {
263                    let _ = RemoteConfigCacheRepo::new(&self.store).put(
264                        REMOTE_CONFIG_RESOURCE,
265                        &raw,
266                        self.clock.now_unix_ms(),
267                    );
268                }
269                {
270                    let mut guard = self.state.write().map_err(|_| RovenueError::Internal())?;
271                    *guard = build_state(body.data);
272                }
273                self.hydrated.store(true, Ordering::SeqCst);
274                self.last_refresh_ms
275                    .store(self.clock.now_unix_ms(), Ordering::SeqCst);
276                if let Some(bus) = &self.bus {
277                    bus.emit(ChangeEvent::RemoteConfigChanged);
278                }
279                Ok(())
280            }
281            Err(e) if matches!(e.kind, ErrorKind::NetworkUnavailable | ErrorKind::Timeout) => {
282                // Connectivity failure: keep serving the last-known payload.
283                // Hydrate from disk if memory is still cold, then surface the
284                // original network error to the caller.
285                self.ensure_hydrated();
286                Err(e)
287            }
288            Err(e) => Err(e),
289        }
290    }
291}
292
293fn build_state(resp: ConfigResponse) -> State {
294    let experiments = resp
295        .experiments
296        .into_iter()
297        .map(|(k, w)| {
298            (
299                k,
300                ExperimentAssignment {
301                    experiment_id: w.experiment_id,
302                    key: w.key,
303                    variant_id: w.variant_id,
304                    variant_name: w.variant_name,
305                    value_json: w.value.to_string(),
306                },
307            )
308        })
309        .collect();
310    State {
311        flags: resp.flags,
312        experiments,
313    }
314}
315
316#[cfg(test)]
317mod tests {
318    use super::*;
319
320    const BODY: &str = r#"{"data":{
321        "flags":{
322            "new_funnel": true,
323            "max_items": 5,
324            "ratio": 1.5,
325            "welcome_text": "hi",
326            "theme": {"color":"blue"}
327        },
328        "experiments":{
329            "checkout_test":{
330                "experimentId":"exp_1",
331                "key":"checkout_test",
332                "type":"FEATURE",
333                "variantId":"var_b",
334                "variantName":"Treatment",
335                "value":{"layout":"compact"}
336            }
337        }
338    }}"#;
339
340    fn make_reader(http: Arc<HttpClient>) -> (RemoteConfigReader, Arc<CacheStore>) {
341        let store = Arc::new(CacheStore::open_in_memory().unwrap());
342        let bus = Arc::new(ObserverBus::default());
343        let clock: Arc<dyn Clock> = Arc::new(SystemClock);
344        let identity = Arc::new(IdentityManager::new(
345            Arc::clone(&store),
346            Arc::clone(&bus),
347            Arc::clone(&clock),
348        ));
349        let reader =
350            RemoteConfigReader::new(http, Arc::clone(&store), identity).with_observer_bus(bus);
351        (reader, store)
352    }
353
354    #[test]
355    #[serial_test::serial]
356    fn refresh_parses_flags_and_experiments_with_typed_getters() {
357        let mut server = mockito::Server::new();
358        let _m = server
359            .mock("GET", "/v1/config")
360            .with_status(200)
361            .with_header("content-type", "application/json")
362            .with_body(BODY)
363            .create();
364
365        let http = Arc::new(HttpClient::new(server.url(), "pk_test".into()));
366        let (reader, _store) = make_reader(http);
367
368        reader.refresh().expect("refresh ok");
369
370        assert!(reader.bool("new_funnel", false));
371        assert_eq!(reader.int("max_items", 0), 5);
372        assert_eq!(reader.double("ratio", 0.0), 1.5);
373        assert_eq!(reader.string("welcome_text", "x".into()), "hi");
374        assert_eq!(reader.json("theme").as_deref(), Some(r#"{"color":"blue"}"#));
375
376        // Wrong-type / unknown keys fall back.
377        assert!(reader.bool("missing", true));
378        assert_eq!(reader.int("welcome_text", -1), -1);
379
380        let exp = reader.experiment("checkout_test").expect("assignment");
381        assert_eq!(exp.experiment_id, "exp_1");
382        assert_eq!(exp.variant_name, "Treatment");
383        assert_eq!(exp.value_json, r#"{"layout":"compact"}"#);
384        assert_eq!(reader.experiments_all().len(), 1);
385        assert!(reader.experiment("nope").is_none());
386
387        // Bulk accessor round-trips to a single parseable object.
388        let all: serde_json::Value = serde_json::from_str(&reader.all_json()).unwrap();
389        assert_eq!(all["flags"]["new_funnel"], serde_json::json!(true));
390        assert_eq!(all["flags"]["max_items"], serde_json::json!(5));
391        assert_eq!(
392            all["experiments"]["checkout_test"]["variantName"],
393            serde_json::json!("Treatment")
394        );
395        assert_eq!(
396            all["experiments"]["checkout_test"]["value"]["layout"],
397            serde_json::json!("compact")
398        );
399    }
400
401    #[test]
402    #[serial_test::serial]
403    fn sends_env_and_subscriber_headers() {
404        let mut server = mockito::Server::new();
405        let m = server
406            .mock("GET", "/v1/config")
407            .match_header("x-rovenue-env", "staging")
408            .match_header("x-rovenue-user-id", mockito::Matcher::Any)
409            .with_status(200)
410            .with_header("content-type", "application/json")
411            .with_body(r#"{"data":{"flags":{},"experiments":{}}}"#)
412            .create();
413
414        let http = Arc::new(
415            HttpClient::new(server.url(), "pk_test".into())
416                .with_environment(Some("staging".into())),
417        );
418        let (reader, _store) = make_reader(http);
419
420        reader.refresh().expect("refresh ok");
421        m.assert();
422    }
423
424    #[test]
425    #[serial_test::serial]
426    fn serves_cached_payload_offline() {
427        // First reader refreshes successfully and persists to the shared store.
428        let mut server = mockito::Server::new();
429        let _m = server
430            .mock("GET", "/v1/config")
431            .with_status(200)
432            .with_header("content-type", "application/json")
433            .with_body(BODY)
434            .create();
435
436        let store = Arc::new(CacheStore::open_in_memory().unwrap());
437        let bus = Arc::new(ObserverBus::default());
438        let clock: Arc<dyn Clock> = Arc::new(SystemClock);
439        let identity = Arc::new(IdentityManager::new(
440            Arc::clone(&store),
441            Arc::clone(&bus),
442            Arc::clone(&clock),
443        ));
444        let http = Arc::new(HttpClient::new(server.url(), "pk_test".into()));
445        let reader = RemoteConfigReader::new(http, Arc::clone(&store), Arc::clone(&identity));
446        reader.refresh().expect("first refresh ok");
447
448        // A brand-new reader over the SAME store, pointed at a dead endpoint,
449        // still resolves flags from the persisted payload.
450        let dead_http = Arc::new(HttpClient::new(
451            "http://127.0.0.1:1".into(),
452            "pk_test".into(),
453        ));
454        let cold = RemoteConfigReader::new(dead_http, Arc::clone(&store), identity);
455        assert!(cold.bool("new_funnel", false));
456        assert_eq!(cold.int("max_items", 0), 5);
457    }
458}
rovenue/remote_config/reader.rs

rovenue/remote_config/
reader.rs
