rovenue/placements/

client.rs

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use crate::cache::placements::PlacementsCacheRepo;
use crate::cache::CacheStore;
use crate::error::{ErrorKind, RovenueError, RovenueResult};
use crate::offerings::client::map_offering;
use crate::time::{Clock, SystemClock};
use crate::transport::api::ApiEnvelope;
use crate::transport::http_client::HttpClient;
use crate::transport::types::HttpRequest;

use super::bucketing::{assign_bucket, select_variant_index};
use super::types::{CoreFunnel, CorePresentedContext, PlacementsResponse};

pub struct PlacementsClient {
    http: Arc<HttpClient>,
    store: Arc<CacheStore>,
    clock: Arc<dyn Clock>,
    /// In-memory bundled-fallback-file map: `identifier -> raw
    /// PlacementsResponse JSON string`, loaded (once, wholesale-replaced)
    /// by `set_fallback` (see `RovenueCore::set_fallback_placements` /
    /// `placements::fallback::parse_fallback_file`). Decoded lazily per
    /// `get_funnel` call — same raw-string-storage rationale as the disk
    /// cache (see `fallback.rs` doc comment): `PlacementsResponse` isn't
    /// `Clone`.
    fallback: Mutex<HashMap<String, String>>,
}

impl PlacementsClient {
    pub fn new(http: Arc<HttpClient>, store: Arc<CacheStore>) -> Self {
        Self {
            http,
            store,
            clock: Arc::new(SystemClock),
            fallback: Mutex::new(HashMap::new()),
        }
    }

    pub fn with_clock(mut self, clock: Arc<dyn Clock>) -> Self {
        self.clock = clock;
        self
    }

    /// Replace the in-memory bundled-fallback-file map wholesale (called
    /// once by `RovenueCore::set_fallback_placements`, after
    /// `fallback::parse_fallback_file` has already validated the file and
    /// decoded each entry once to confirm it parses as `PlacementsResponse`
    /// — the raw strings stored here are re-decoded lazily per
    /// `get_funnel` call, same as the disk cache).
    pub fn set_fallback(&self, entries: HashMap<String, String>) {
        if let Ok(mut guard) = self.fallback.lock() {
            *guard = entries;
        }
    }

    /// Fetch + resolve `GET /v1/placements/{identifier}?locale=` into the
    /// funnel the placement serves.
    ///
    /// `Ok(None)` means the placement resolved to nothing — unknown/inactive
    /// placement, a `target: none` row, a retired reference, or no row
    /// matched. This is NOT an error: a shipped app must never crash
    /// because a placement was retired server-side.
    ///
    /// On a successful live fetch the raw response JSON is cached under
    /// `placement:{identifier}`. On `NetworkUnavailable`/`Timeout` the last
    /// cached response (if any) is re-resolved instead (a fresh bucket draw
    /// runs against the cached experiment payload). If there's no cached
    /// response either, the bundled fallback map (see `set_fallback`) is
    /// consulted as a last resort — cache always beats fallback, and the
    /// fallback is only ever consulted on this same connectivity-class
    /// branch (never for auth/server errors, which propagate exactly as
    /// before). Any other error propagates unchanged.
    ///
    /// `server_assignment(experiment_key)` returns the variant the server
    /// already assigned this subscriber (the Remote Config cache), if any.
    /// It wins over the local draw, which otherwise disagrees with the
    /// server after two devices merge into one subscriber (see
    /// `resolve_funnel`).
    pub fn get_funnel(
        &self,
        identifier: &str,
        locale: Option<&str>,
        subscriber_id: &str,
        server_assignment: &dyn Fn(&str) -> Option<String>,
    ) -> RovenueResult<Option<CoreFunnel>> {
        let (resp, served_from_fallback) = self.fetch(identifier, locale, subscriber_id)?;
        Ok(resolve_funnel(
            resp,
            subscriber_id,
            served_from_fallback,
            server_assignment,
        ))
    }

    /// The shared fetch ladder: live, then disk cache, then the bundled
    /// fallback file. The bool is `served_from_fallback`.
    fn fetch(
        &self,
        identifier: &str,
        locale: Option<&str>,
        subscriber_id: &str,
    ) -> RovenueResult<(PlacementsResponse, bool)> {
        let cache_key = format!("placement:{identifier}");
        let path = match locale {
            Some(l) if !l.is_empty() => format!("/v1/placements/{identifier}?locale={l}"),
            _ => format!("/v1/placements/{identifier}"),
        };
        let req = HttpRequest::new(&path).subscriber_id(subscriber_id);

        match self.http.get_json::<ApiEnvelope<PlacementsResponse>>(req) {
            Ok(resp) => {
                let body = resp.body.ok_or(RovenueError::Internal())?;
                // Persist the raw response (best-effort: a cache write
                // failure must not fail an otherwise-successful fetch).
                if let Ok(raw) = serde_json::to_string(&body.data) {
                    let _ = PlacementsCacheRepo::new(&self.store).put(
                        &cache_key,
                        &raw,
                        self.clock.now_unix_ms(),
                    );
                }
                Ok((body.data, false))
            }
            Err(e) if matches!(e.kind, ErrorKind::NetworkUnavailable | ErrorKind::Timeout) => {
                match PlacementsCacheRepo::new(&self.store).get(&cache_key)? {
                    Some(raw) => {
                        let parsed: PlacementsResponse =
                            serde_json::from_str(&raw).map_err(|_| RovenueError::Internal())?;
                        Ok((parsed, false))
                    }
                    None => match self.fallback_raw(identifier) {
                        Some(raw) => match serde_json::from_str::<PlacementsResponse>(&raw) {
                            Ok(parsed) => Ok((parsed, true)),
                            // A stored fallback entry that fails to
                            // re-decode is unreachable in practice (it was
                            // already validated by parse_fallback_file
                            // before being stored), but must not panic —
                            // fall through to the original network error.
                            Err(_) => Err(e),
                        },
                        None => Err(e),
                    },
                }
            }
            Err(e) => Err(e),
        }
    }

    fn fallback_raw(&self, identifier: &str) -> Option<String> {
        self.fallback
            .lock()
            .ok()
            .and_then(|guard| guard.get(identifier).cloned())
    }
}

/// Pull a funnel out of a resolved placement response.
///
/// A free function rather than a method: it touches no client state, and
/// keeping it that way makes it testable without an HTTP stack.
fn resolve_funnel(
    resp: PlacementsResponse,
    subscriber_id: &str,
    served_from_fallback: bool,
    server_assignment: &dyn Fn(&str) -> Option<String>,
) -> Option<CoreFunnel> {
    let placement = resp.placement?;

    // A direct target wins; otherwise draw a variant.
    let (funnel, variant_id, experiment_key) = match resp.funnel {
        Some(funnel) => (funnel, None, None),
        None => {
            let experiment = resp.experiment?;
            if experiment.variants.is_empty() {
                return None;
            }
            // The server's sticky assignment first: after two devices merge
            // into one subscriber, this device's rovenueId can draw a
            // different variant than the one the server recorded (drawn
            // from the other device's rovenueId). /v1/config is per
            // subscriber and follows the merge, so its cached assignment is
            // the canonical one. A variant the envelope no longer carries
            // (the experiment was edited) falls through to the draw.
            let assigned = server_assignment(&experiment.key)
                .and_then(|id| experiment.variants.iter().position(|v| v.variant_id == id));
            // Otherwise the local draw: one bucketing function, one set of
            // vectors, one weight interpretation — shared with every other
            // experiment draw in this crate, so all of them agree with
            // bucketing-vectors.json. For a single device it lands where the
            // server's own draw did.
            let idx = assigned.unwrap_or_else(|| {
                let weights: Vec<f64> = experiment.variants.iter().map(|v| v.weight).collect();
                select_variant_index(assign_bucket(subscriber_id, &experiment.key), &weights)
            });
            let variant = experiment.variants.into_iter().nth(idx)?;
            (
                variant.funnel,
                Some(variant.variant_id),
                Some(experiment.key),
            )
        }
    };
    // Serialized here rather than on the way in, so the string the façade
    // receives is exactly what `FunnelSession` parses.
    let config_json = serde_json::to_string(&funnel.config).ok()?;

    Some(CoreFunnel {
        placement_identifier: placement.identifier.clone(),
        placement_revision: placement.revision,
        funnel_identifier: funnel.identifier.clone(),
        funnel_id: funnel.id.clone(),
        revision: funnel.revision,
        config_format_version: funnel.config_format_version,
        config_json,
        offering: funnel.offering.map(map_offering),
        links: funnel.links.into(),
        // `funnel_id` holds the funnels ROW id — what web checkout writes
        // too (funnel-payment.ts) — not the document identifier.
        presented_context: Some(CorePresentedContext {
            placement_id: placement.identifier,
            funnel_id: funnel.id.clone(),
            variant_id,
            experiment_key,
            revision: placement.revision,
        }),
        served_from_fallback,
    })
}